1package coordinator
  2
  3import (
  4	"context"
  5	"fmt"
  6	"time"
  7
  8	ethCommon "github.com/ethereum/go-ethereum/common"
  9	"github.com/ethereum/go-ethereum/core/types"
 10	"github.com/hermeznetwork/hermez-node/batchbuilder"
 11	"github.com/hermeznetwork/hermez-node/common"
 12	"github.com/hermeznetwork/hermez-node/db/historydb"
 13	"github.com/hermeznetwork/hermez-node/eth"
 14	"github.com/hermeznetwork/hermez-node/log"
 15	"github.com/hermeznetwork/hermez-node/txselector"
 16)
 17
 18var errTODO = fmt.Errorf("TODO")
 19
 20// ErrStop is returned when the function is stopped asynchronously via the stop
 21// channel.  It doesn't indicate an error.
 22var ErrStop = fmt.Errorf("Stopped")
 23
 24// Config contains the Coordinator configuration
 25type Config struct {
 26	ForgerAddress ethCommon.Address
 27}
 28
 29// Coordinator implements the Coordinator type
 30type Coordinator struct {
 31	forging bool
 32	// rw         *sync.RWMutex
 33	// isForgeSeq bool // WIP just for testing while implementing
 34
 35	config Config
 36
 37	batchNum        common.BatchNum
 38	serverProofPool *ServerProofPool
 39
 40	// synchronizer *synchronizer.Synchronizer
 41	hdb          *historydb.HistoryDB
 42	txsel        *txselector.TxSelector
 43	batchBuilder *batchbuilder.BatchBuilder
 44
 45	ethClient eth.ClientInterface
 46	ethTxs    []*types.Transaction
 47	// ethTxStore kvdb.Storage
 48}
 49
 50// NewCoordinator creates a new Coordinator
 51func NewCoordinator(conf Config,
 52	hdb *historydb.HistoryDB,
 53	txsel *txselector.TxSelector,
 54	bb *batchbuilder.BatchBuilder,
 55	serverProofs []ServerProofInterface,
 56	ethClient eth.ClientInterface) *Coordinator { // once synchronizer is ready, synchronizer.Synchronizer will be passed as parameter here
 57	serverProofPool := NewServerProofPool(len(serverProofs))
 58	for _, serverProof := range serverProofs {
 59		serverProofPool.Add(serverProof)
 60	}
 61	c := Coordinator{
 62		config:          conf,
 63		serverProofPool: serverProofPool,
 64		hdb:             hdb,
 65		txsel:           txsel,
 66		batchBuilder:    bb,
 67		ethClient:       ethClient,
 68
 69		ethTxs: make([]*types.Transaction, 0),
 70		// ethTxStore:      memory.NewMemoryStorage(),
 71		// rw:              &sync.RWMutex{},
 72	}
 73	return &c
 74}
 75
 76// TODO(Edu): Change the current design of the coordinator structur:
 77// - Move Start and Stop functions (from node/node.go) here
 78// - Add concept of StartPipeline, StopPipeline, that spawns and stops the goroutines
 79// - Add a Manager that calls StartPipeline and StopPipeline, checks when it's time to forge, schedules new batches, etc.
 80// - Add a TxMonitor that monitors successful ForgeBatch ethereum transactions and waits for N blocks of confirmation, and reports back errors to the Manager.
 81
 82// ForgeLoopFn is the function ran in a loop that checks if it's time to forge
 83// and forges a batch if so and sends it to outBatchCh.  Returns true if it's
 84// the coordinator turn to forge.
 85func (c *Coordinator) ForgeLoopFn(outBatchCh chan *BatchInfo, stopCh chan bool) (forgetime bool, err error) {
 86	// TODO: Move the logic to check if it's forge time or not outside the pipeline
 87	isForgeSequence, err := c.isForgeSequence()
 88	if err != nil {
 89		return false, err
 90	}
 91	if !isForgeSequence {
 92		if c.forging {
 93			log.Info("ForgeLoopFn: forging state end")
 94			c.forging = false
 95		}
 96		log.Debug("ForgeLoopFn: not in forge time")
 97		return false, nil
 98	}
 99	log.Debug("ForgeLoopFn: forge time")
100	if !c.forging {
101		// Start pipeline from a batchNum state taken from synchronizer
102		log.Info("ForgeLoopFn: forging state begin")
103		// c.batchNum = c.hdb.GetLastBatchNum() // uncomment when HistoryDB is ready
104		err := c.txsel.Reset(c.batchNum)
105		if err != nil {
106			log.Errorw("ForgeLoopFn: TxSelector.Reset", "error", err)
107			return true, err
108		}
109		err = c.batchBuilder.Reset(c.batchNum, true)
110		if err != nil {
111			log.Errorw("ForgeLoopFn: BatchBuilder.Reset", "error", err)
112			return true, err
113		}
114		// c.batchQueue = NewBatchQueue()
115		c.forging = true
116	}
117	// TODO once synchronizer has this method ready:
118	// If there's been a reorg, handle it
119	// handleReorg() function decides if the reorg must restart the pipeline or not
120	// if c.synchronizer.Reorg():
121	_ = c.handleReorg()
122
123	defer func() {
124		if err == ErrStop {
125			log.Info("ForgeLoopFn: forgeLoopFn stopped")
126		}
127	}()
128
129	// 0. Wait for an available server proof
130	// blocking call
131	serverProof, err := c.serverProofPool.Get(stopCh)
132	if err != nil {
133		return true, err
134	}
135	defer func() {
136		if !forgetime || err != nil {
137			c.serverProofPool.Add(serverProof)
138		}
139	}()
140
141	log.Debugw("ForgeLoopFn: using serverProof", "server", serverProof)
142	log.Debugw("ForgeLoopFn: forge start")
143	// forge for batchNum = batchNum + 1.
144	batchInfo, err := c.forge(serverProof)
145	if err != nil {
146		log.Errorw("forge", "error", err)
147		return true, err
148	}
149	log.Debugw("ForgeLoopFn: forge end", "batchNum", batchInfo.batchNum)
150	outBatchCh <- batchInfo
151	return true, nil
152}
153
154// GetProofCallForgeLoopFn is the function ran in a loop that gets a forged
155// batch via inBatchCh, waits for the proof server to finish, calls the ForgeBatch
156// function in the Rollup Smart Contract, and sends the batch to outBatchCh.
157func (c *Coordinator) GetProofCallForgeLoopFn(inBatchCh, outBatchCh chan *BatchInfo, stopCh chan bool) (err error) {
158	defer func() {
159		if err == ErrStop {
160			log.Info("GetProofCallForgeLoopFn: forgeLoopFn stopped")
161		}
162	}()
163	select {
164	case <-stopCh:
165		return ErrStop
166	case batchInfo := <-inBatchCh:
167		log.Debugw("GetProofCallForgeLoopFn: getProofCallForge start", "batchNum", batchInfo.batchNum)
168		if err := c.getProofCallForge(batchInfo, stopCh); err != nil {
169			return err
170		}
171		log.Debugw("GetProofCallForgeLoopFn: getProofCallForge end", "batchNum", batchInfo.batchNum)
172		outBatchCh <- batchInfo
173	}
174	return nil
175}
176
177// ForgeCallConfirmLoopFn is the function ran in a loop that gets a batch that
178// has been sent to the Rollup Smart Contract via inBatchCh and waits for the
179// ethereum transaction confirmation.
180func (c *Coordinator) ForgeCallConfirmLoopFn(inBatchCh chan *BatchInfo, stopCh chan bool) (err error) {
181	defer func() {
182		if err == ErrStop {
183			log.Info("ForgeCallConfirmLoopFn: forgeConfirmLoopFn stopped")
184		}
185	}()
186	select {
187	case <-stopCh:
188		return ErrStop
189	case batchInfo := <-inBatchCh:
190		log.Debugw("ForgeCallConfirmLoopFn: forgeCallConfirm start", "batchNum", batchInfo.batchNum)
191		if err := c.forgeCallConfirm(batchInfo, stopCh); err != nil {
192			return err
193		}
194		log.Debugw("ForgeCallConfirmLoopFn: forgeCallConfirm  end", "batchNum", batchInfo.batchNum)
195	}
196	return nil
197}
198
199func (c *Coordinator) forge(serverProof ServerProofInterface) (*BatchInfo, error) {
200	// remove transactions from the pool that have been there for too long
201	err := c.purgeRemoveByTimeout()
202	if err != nil {
203		return nil, err
204	}
205
206	c.batchNum = c.batchNum + 1
207	batchInfo := NewBatchInfo(c.batchNum, serverProof) // to accumulate metadata of the batch
208
209	var poolL2Txs []common.PoolL2Tx
210	// var feesInfo
211	var l1UserTxsExtra, l1OperatorTxs []common.L1Tx
212	// 1. Decide if we forge L2Tx or L1+L2Tx
213	if c.shouldL1L2Batch() {
214		// 2a: L1+L2 txs
215		// l1UserTxs, toForgeL1TxsNumber := c.hdb.GetNextL1UserTxs() // TODO once HistoryDB is ready, uncomment
216		var l1UserTxs []common.L1Tx = nil                                                                                 // tmp, depends on HistoryDB
217		l1UserTxsExtra, l1OperatorTxs, poolL2Txs, err = c.txsel.GetL1L2TxSelection([]common.Idx{}, c.batchNum, l1UserTxs) // TODO once feesInfo is added to method return, add the var
218		if err != nil {
219			return nil, err
220		}
221	} else {
222		// 2b: only L2 txs
223		poolL2Txs, err = c.txsel.GetL2TxSelection([]common.Idx{}, c.batchNum) // TODO once feesInfo is added to method return, add the var
224		if err != nil {
225			return nil, err
226		}
227		l1UserTxsExtra = nil
228		l1OperatorTxs = nil
229	}
230
231	// Run purger to invalidate transactions that become invalid beause of
232	// the poolL2Txs selected.  Will mark as invalid the txs that have a
233	// (fromIdx, nonce) which already appears in the selected txs (includes
234	// all the nonces smaller than the current one)
235	err = c.purgeInvalidDueToL2TxsSelection(poolL2Txs)
236	if err != nil {
237		return nil, err
238	}
239
240	// 3.  Save metadata from TxSelector output for BatchNum
241	batchInfo.SetTxsInfo(l1UserTxsExtra, l1OperatorTxs, poolL2Txs) // TODO feesInfo
242
243	// 4. Call BatchBuilder with TxSelector output
244	configBatch := &batchbuilder.ConfigBatch{
245		ForgerAddress: c.config.ForgerAddress,
246	}
247	zkInputs, err := c.batchBuilder.BuildBatch([]common.Idx{}, configBatch, l1UserTxsExtra, l1OperatorTxs, poolL2Txs, nil) // TODO []common.TokenID --> feesInfo
248	if err != nil {
249		return nil, err
250	}
251
252	// 5. Save metadata from BatchBuilder output for BatchNum
253	batchInfo.SetZKInputs(zkInputs)
254
255	// 6. Call an idle server proof with BatchBuilder output, save server proof info for batchNum
256	err = batchInfo.serverProof.CalculateProof(zkInputs)
257	if err != nil {
258		return nil, err
259	}
260
261	return &batchInfo, nil
262}
263
264// getProofCallForge gets the generated zkProof & sends it to the SmartContract
265func (c *Coordinator) getProofCallForge(batchInfo *BatchInfo, stopCh chan bool) error {
266	serverProof := batchInfo.serverProof
267	proof, err := serverProof.GetProof(stopCh) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
268	c.serverProofPool.Add(serverProof)
269	batchInfo.serverProof = nil
270	if err != nil {
271		return err
272	}
273	batchInfo.SetProof(proof)
274	forgeBatchArgs := c.prepareForgeBatchArgs(batchInfo)
275	ethTx, err := c.ethClient.RollupForgeBatch(forgeBatchArgs)
276	if err != nil {
277		return err
278	}
279	// TODO: Move this to the next step (forgeCallConfirm)
280	log.Debugf("ethClient ForgeCall sent, batchNum: %d", c.batchNum)
281	batchInfo.SetEthTx(ethTx)
282
283	// TODO(FUTURE) once tx data type is defined, store ethTx (returned by ForgeCall)
284	// TBD if use ethTxStore as a disk k-v database, or use a Queue
285	// tx, err := c.ethTxStore.NewTx()
286	// if err != nil {
287	//         return err
288	// }
289	// tx.Put(ethTx.Hash(), ethTx.Bytes())
290	// if err := tx.Commit(); err!=nil {
291	//         return nil
292	// }
293
294	return nil
295}
296
297func (c *Coordinator) forgeCallConfirm(batchInfo *BatchInfo, stopCh chan bool) error {
298	// TODO strategy of this sequence TBD
299	// confirm eth txs and mark them as accepted sequence
300	// IDEA: Keep an array in Coordinator with the list of sent ethTx.
301	// Here, loop over them and only delete them once the number of
302	// confirmed blocks is over a configured value.  If the tx is rejected,
303	// return error.
304	// ethTx := ethTxStore.GetFirstPending()
305	// waitForAccepted(ethTx) // blocking call, returns once the ethTx is mined
306	// ethTxStore.MarkAccepted(ethTx)
307	txID := batchInfo.ethTx.Hash()
308	// TODO: Follow EthereumClient.waitReceipt logic
309	count := 0
310	// TODO: Define this waitTime in the config
311	waitTime := 100 * time.Millisecond //nolint:gomnd
312	select {
313	case <-time.After(waitTime):
314		receipt, err := c.ethClient.EthTransactionReceipt(context.TODO(), txID)
315		if err != nil {
316			return err
317		}
318		if receipt != nil {
319			if receipt.Status == types.ReceiptStatusFailed {
320				return fmt.Errorf("receipt status is failed")
321			} else if receipt.Status == types.ReceiptStatusSuccessful {
322				return nil
323			}
324		}
325		// TODO: Call go-ethereum:
326		// if err == nil && receipt == nil :
327		// `func (ec *Client) TransactionByHash(ctx context.Context, hash common.Hash) (tx *types.Transaction, isPending bool, err error) {`
328		count++
329		if time.Duration(count)*waitTime > 60*time.Second {
330			log.Warnw("Waiting for ethTx receipt for more than 60 seconds", "tx", batchInfo.ethTx)
331			// TODO: Decide if we resend the Tx with higher gas price
332		}
333	case <-stopCh:
334		return ErrStop
335	}
336	return fmt.Errorf("timeout")
337}
338
339func (c *Coordinator) handleReorg() error {
340	return nil // TODO
341}
342
343// isForgeSequence returns true if the node is the Forger in the current ethereum block
344func (c *Coordinator) isForgeSequence() (bool, error) {
345	// TODO: Consider checking if we can forge by quering the Synchronizer instead of using ethClient
346	blockNum, err := c.ethClient.EthCurrentBlock()
347	if err != nil {
348		return false, err
349	}
350	addr, err := c.ethClient.EthAddress()
351	if err != nil {
352		return false, err
353	}
354	return c.ethClient.AuctionCanForge(*addr, blockNum+1)
355}
356
357func (c *Coordinator) purgeRemoveByTimeout() error {
358	return nil // TODO
359}
360
361func (c *Coordinator) purgeInvalidDueToL2TxsSelection(l2Txs []common.PoolL2Tx) error {
362	return nil // TODO
363}
364
365func (c *Coordinator) shouldL1L2Batch() bool {
366	return false // TODO
367}
368
369func (c *Coordinator) prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
370	// TODO
371	return &eth.RollupForgeBatchArgs{}
372}