hermez-node/coordinator/coordinator.go

package coordinator

import (
	"context"
	"fmt"
	"time"

	ethCommon "github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/hermeznetwork/hermez-node/batchbuilder"
	"github.com/hermeznetwork/hermez-node/common"
	"github.com/hermeznetwork/hermez-node/db/historydb"
	"github.com/hermeznetwork/hermez-node/eth"
	"github.com/hermeznetwork/hermez-node/log"
	"github.com/hermeznetwork/hermez-node/txselector"
)

var errTODO = fmt.Errorf("TODO")

// ErrStop is returned when the function is stopped asynchronously via the stop
// channel.  It doesn't indicate an error.
var ErrStop = fmt.Errorf("Stopped")

// Config contains the Coordinator configuration
type Config struct {
	ForgerAddress ethCommon.Address
}

// Coordinator implements the Coordinator type
type Coordinator struct {
	forging bool
	// rw         *sync.RWMutex
	// isForgeSeq bool // WIP just for testing while implementing

	config Config

	batchNum        common.BatchNum
	serverProofPool *ServerProofPool

	// synchronizer *synchronizer.Synchronizer
	hdb          *historydb.HistoryDB
	txsel        *txselector.TxSelector
	batchBuilder *batchbuilder.BatchBuilder

	ethClient eth.ClientInterface
	ethTxs    []*types.Transaction
	// ethTxStore kvdb.Storage
}

// NewCoordinator creates a new Coordinator
func NewCoordinator(conf Config,
	hdb *historydb.HistoryDB,
	txsel *txselector.TxSelector,
	bb *batchbuilder.BatchBuilder,
	serverProofs []ServerProofInterface,
	ethClient eth.ClientInterface) *Coordinator { // once synchronizer is ready, synchronizer.Synchronizer will be passed as parameter here
	serverProofPool := NewServerProofPool(len(serverProofs))
	for _, serverProof := range serverProofs {
		serverProofPool.Add(serverProof)
	}
	c := Coordinator{
		config:          conf,
		serverProofPool: serverProofPool,
		hdb:             hdb,
		txsel:           txsel,
		batchBuilder:    bb,
		ethClient:       ethClient,

		ethTxs: make([]*types.Transaction, 0),
		// ethTxStore:      memory.NewMemoryStorage(),
		// rw:              &sync.RWMutex{},
	}
	return &c
}

// TODO(Edu): Change the current design of the coordinator structur:
// - Move Start and Stop functions (from node/node.go) here
// - Add concept of StartPipeline, StopPipeline, that spawns and stops the goroutines
// - Add a Manager that calls StartPipeline and StopPipeline, checks when it's time to forge, schedules new batches, etc.
// - Add a TxMonitor that monitors successful ForgeBatch ethereum transactions and waits for N blocks of confirmation, and reports back errors to the Manager.

// ForgeLoopFn is the function ran in a loop that checks if it's time to forge
// and forges a batch if so and sends it to outBatchCh.  Returns true if it's
// the coordinator turn to forge.
func (c *Coordinator) ForgeLoopFn(outBatchCh chan *BatchInfo, stopCh chan bool) (forgetime bool, err error) {
	// TODO: Move the logic to check if it's forge time or not outside the pipeline
	isForgeSequence, err := c.isForgeSequence()
	if err != nil {
		return false, err
	}
	if !isForgeSequence {
		if c.forging {
			log.Info("ForgeLoopFn: forging state end")
			c.forging = false
		}
		log.Debug("ForgeLoopFn: not in forge time")
		return false, nil
	}
	log.Debug("ForgeLoopFn: forge time")
	if !c.forging {
		// Start pipeline from a batchNum state taken from synchronizer
		log.Info("ForgeLoopFn: forging state begin")
		// c.batchNum = c.hdb.GetLastBatchNum() // uncomment when HistoryDB is ready
		err := c.txsel.Reset(c.batchNum)
		if err != nil {
			log.Errorw("ForgeLoopFn: TxSelector.Reset", "error", err)
			return true, err
		}
		err = c.batchBuilder.Reset(c.batchNum, true)
		if err != nil {
			log.Errorw("ForgeLoopFn: BatchBuilder.Reset", "error", err)
			return true, err
		}
		// c.batchQueue = NewBatchQueue()
		c.forging = true
	}
	// TODO once synchronizer has this method ready:
	// If there's been a reorg, handle it
	// handleReorg() function decides if the reorg must restart the pipeline or not
	// if c.synchronizer.Reorg():
	_ = c.handleReorg()

	defer func() {
		if err == ErrStop {
			log.Info("ForgeLoopFn: forgeLoopFn stopped")
		}
	}()

	// 0. Wait for an available server proof
	// blocking call
	serverProof, err := c.serverProofPool.Get(stopCh)
	if err != nil {
		return true, err
	}
	defer func() {
		if !forgetime || err != nil {
			c.serverProofPool.Add(serverProof)
		}
	}()

	log.Debugw("ForgeLoopFn: using serverProof", "server", serverProof)
	log.Debugw("ForgeLoopFn: forge start")
	// forge for batchNum = batchNum + 1.
	batchInfo, err := c.forge(serverProof)
	if err != nil {
		log.Errorw("forge", "error", err)
		return true, err
	}
	log.Debugw("ForgeLoopFn: forge end", "batchNum", batchInfo.batchNum)
	outBatchCh <- batchInfo
	return true, nil
}

// GetProofCallForgeLoopFn is the function ran in a loop that gets a forged
// batch via inBatchCh, waits for the proof server to finish, calls the ForgeBatch
// function in the Rollup Smart Contract, and sends the batch to outBatchCh.
func (c *Coordinator) GetProofCallForgeLoopFn(inBatchCh, outBatchCh chan *BatchInfo, stopCh chan bool) (err error) {
	defer func() {
		if err == ErrStop {
			log.Info("GetProofCallForgeLoopFn: forgeLoopFn stopped")
		}
	}()
	select {
	case <-stopCh:
		return ErrStop
	case batchInfo := <-inBatchCh:
		log.Debugw("GetProofCallForgeLoopFn: getProofCallForge start", "batchNum", batchInfo.batchNum)
		if err := c.getProofCallForge(batchInfo, stopCh); err != nil {
			return err
		}
		log.Debugw("GetProofCallForgeLoopFn: getProofCallForge end", "batchNum", batchInfo.batchNum)
		outBatchCh <- batchInfo
	}
	return nil
}

// ForgeCallConfirmLoopFn is the function ran in a loop that gets a batch that
// has been sent to the Rollup Smart Contract via inBatchCh and waits for the
// ethereum transaction confirmation.
func (c *Coordinator) ForgeCallConfirmLoopFn(inBatchCh chan *BatchInfo, stopCh chan bool) (err error) {
	defer func() {
		if err == ErrStop {
			log.Info("ForgeCallConfirmLoopFn: forgeConfirmLoopFn stopped")
		}
	}()
	select {
	case <-stopCh:
		return ErrStop
	case batchInfo := <-inBatchCh:
		log.Debugw("ForgeCallConfirmLoopFn: forgeCallConfirm start", "batchNum", batchInfo.batchNum)
		if err := c.forgeCallConfirm(batchInfo, stopCh); err != nil {
			return err
		}
		log.Debugw("ForgeCallConfirmLoopFn: forgeCallConfirm  end", "batchNum", batchInfo.batchNum)
	}
	return nil
}

func (c *Coordinator) forge(serverProof ServerProofInterface) (*BatchInfo, error) {
	// remove transactions from the pool that have been there for too long
	err := c.purgeRemoveByTimeout()
	if err != nil {
		return nil, err
	}

	c.batchNum = c.batchNum + 1
	batchInfo := NewBatchInfo(c.batchNum, serverProof) // to accumulate metadata of the batch

	var poolL2Txs []common.PoolL2Tx
	// var feesInfo
	var l1UserTxsExtra, l1OperatorTxs []common.L1Tx
	// 1. Decide if we forge L2Tx or L1+L2Tx
	if c.shouldL1L2Batch() {
		// 2a: L1+L2 txs
		// l1UserTxs, toForgeL1TxsNumber := c.hdb.GetNextL1UserTxs() // TODO once HistoryDB is ready, uncomment
		var l1UserTxs []common.L1Tx = nil                                                                 // tmp, depends on HistoryDB
		l1UserTxsExtra, l1OperatorTxs, poolL2Txs, err = c.txsel.GetL1L2TxSelection(c.batchNum, l1UserTxs) // TODO once feesInfo is added to method return, add the var
		if err != nil {
			return nil, err
		}
	} else {
		// 2b: only L2 txs
		poolL2Txs, err = c.txsel.GetL2TxSelection(c.batchNum) // TODO once feesInfo is added to method return, add the var
		if err != nil {
			return nil, err
		}
		l1UserTxsExtra = nil
		l1OperatorTxs = nil
	}

	// Run purger to invalidate transactions that become invalid beause of
	// the poolL2Txs selected.  Will mark as invalid the txs that have a
	// (fromIdx, nonce) which already appears in the selected txs (includes
	// all the nonces smaller than the current one)
	err = c.purgeInvalidDueToL2TxsSelection(poolL2Txs)
	if err != nil {
		return nil, err
	}

	// 3.  Save metadata from TxSelector output for BatchNum
	batchInfo.SetTxsInfo(l1UserTxsExtra, l1OperatorTxs, poolL2Txs) // TODO feesInfo

	// 4. Call BatchBuilder with TxSelector output
	configBatch := &batchbuilder.ConfigBatch{
		ForgerAddress: c.config.ForgerAddress,
	}
	zkInputs, err := c.batchBuilder.BuildBatch(configBatch, l1UserTxsExtra, l1OperatorTxs, poolL2Txs, nil) // TODO []common.TokenID --> feesInfo
	if err != nil {
		return nil, err
	}

	// 5. Save metadata from BatchBuilder output for BatchNum
	batchInfo.SetZKInputs(zkInputs)

	// 6. Call an idle server proof with BatchBuilder output, save server proof info for batchNum
	err = batchInfo.serverProof.CalculateProof(zkInputs)
	if err != nil {
		return nil, err
	}

	return &batchInfo, nil
}

// getProofCallForge gets the generated zkProof & sends it to the SmartContract
func (c *Coordinator) getProofCallForge(batchInfo *BatchInfo, stopCh chan bool) error {
	serverProof := batchInfo.serverProof
	proof, err := serverProof.GetProof(stopCh) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
	c.serverProofPool.Add(serverProof)
	batchInfo.serverProof = nil
	if err != nil {
		return err
	}
	batchInfo.SetProof(proof)
	forgeBatchArgs := c.prepareForgeBatchArgs(batchInfo)
	ethTx, err := c.ethClient.RollupForgeBatch(forgeBatchArgs)
	if err != nil {
		return err
	}
	// TODO: Move this to the next step (forgeCallConfirm)
	log.Debugf("ethClient ForgeCall sent, batchNum: %d", c.batchNum)
	batchInfo.SetEthTx(ethTx)

	// TODO(FUTURE) once tx data type is defined, store ethTx (returned by ForgeCall)
	// TBD if use ethTxStore as a disk k-v database, or use a Queue
	// tx, err := c.ethTxStore.NewTx()
	// if err != nil {
	//         return err
	// }
	// tx.Put(ethTx.Hash(), ethTx.Bytes())
	// if err := tx.Commit(); err!=nil {
	//         return nil
	// }

	return nil
}

func (c *Coordinator) forgeCallConfirm(batchInfo *BatchInfo, stopCh chan bool) error {
	// TODO strategy of this sequence TBD
	// confirm eth txs and mark them as accepted sequence
	// IDEA: Keep an array in Coordinator with the list of sent ethTx.
	// Here, loop over them and only delete them once the number of
	// confirmed blocks is over a configured value.  If the tx is rejected,
	// return error.
	// ethTx := ethTxStore.GetFirstPending()
	// waitForAccepted(ethTx) // blocking call, returns once the ethTx is mined
	// ethTxStore.MarkAccepted(ethTx)
	txID := batchInfo.ethTx.Hash()
	// TODO: Follow EthereumClient.waitReceipt logic
	count := 0
	// TODO: Define this waitTime in the config
	waitTime := 100 * time.Millisecond //nolint:gomnd
	select {
	case <-time.After(waitTime):
		receipt, err := c.ethClient.EthTransactionReceipt(context.TODO(), txID)
		if err != nil {
			return err
		}
		if receipt != nil {
			if receipt.Status == types.ReceiptStatusFailed {
				return fmt.Errorf("receipt status is failed")
			} else if receipt.Status == types.ReceiptStatusSuccessful {
				return nil
			}
		}
		// TODO: Call go-ethereum:
		// if err == nil && receipt == nil :
		// `func (ec *Client) TransactionByHash(ctx context.Context, hash common.Hash) (tx *types.Transaction, isPending bool, err error) {`
		count++
		if time.Duration(count)*waitTime > 60*time.Second {
			log.Warnw("Waiting for ethTx receipt for more than 60 seconds", "tx", batchInfo.ethTx)
			// TODO: Decide if we resend the Tx with higher gas price
		}
	case <-stopCh:
		return ErrStop
	}
	return fmt.Errorf("timeout")
}

func (c *Coordinator) handleReorg() error {
	return nil // TODO
}

// isForgeSequence returns true if the node is the Forger in the current ethereum block
func (c *Coordinator) isForgeSequence() (bool, error) {
	// TODO: Consider checking if we can forge by quering the Synchronizer instead of using ethClient
	blockNum, err := c.ethClient.EthCurrentBlock()
	if err != nil {
		return false, err
	}
	addr, err := c.ethClient.EthAddress()
	if err != nil {
		return false, err
	}
	return c.ethClient.AuctionCanForge(*addr, blockNum+1)
}

func (c *Coordinator) purgeRemoveByTimeout() error {
	return nil // TODO
}

func (c *Coordinator) purgeInvalidDueToL2TxsSelection(l2Txs []common.PoolL2Tx) error {
	return nil // TODO
}

func (c *Coordinator) shouldL1L2Batch() bool {
	return false // TODO
}

func (c *Coordinator) prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
	// TODO
	return &eth.RollupForgeBatchArgs{}
}