1package coordinator
  2
  3import (
  4	"fmt"
  5
  6	ethCommon "github.com/ethereum/go-ethereum/common"
  7	"github.com/hermeznetwork/hermez-node/batchbuilder"
  8	"github.com/hermeznetwork/hermez-node/common"
  9	"github.com/hermeznetwork/hermez-node/db/historydb"
 10	"github.com/hermeznetwork/hermez-node/eth"
 11	"github.com/hermeznetwork/hermez-node/log"
 12	"github.com/hermeznetwork/hermez-node/txselector"
 13	kvdb "github.com/iden3/go-merkletree/db"
 14	"github.com/iden3/go-merkletree/db/memory"
 15)
 16
 17// ErrStop is returned when the function is stopped asynchronously via the stop
 18// channel.  It doesn't indicate an error.
 19var ErrStop = fmt.Errorf("Stopped")
 20
 21// Config contains the Coordinator configuration
 22type Config struct {
 23	ForgerAddress ethCommon.Address
 24}
 25
 26// Coordinator implements the Coordinator type
 27type Coordinator struct {
 28	forging    bool
 29	isForgeSeq bool // WIP just for testing while implementing
 30
 31	config Config
 32
 33	batchNum        common.BatchNum
 34	serverProofPool *ServerProofPool
 35
 36	// synchronizer *synchronizer.Synchronizer
 37	hdb          *historydb.HistoryDB
 38	txsel        *txselector.TxSelector
 39	batchBuilder *batchbuilder.BatchBuilder
 40
 41	ethClient  eth.ClientInterface
 42	ethTxStore kvdb.Storage
 43}
 44
 45// NewCoordinator creates a new Coordinator
 46func NewCoordinator(conf Config,
 47	hdb *historydb.HistoryDB,
 48	txsel *txselector.TxSelector,
 49	bb *batchbuilder.BatchBuilder,
 50	serverProofs []ServerProofInterface,
 51	ethClient *eth.Client) *Coordinator { // once synchronizer is ready, synchronizer.Synchronizer will be passed as parameter here
 52	serverProofPool := NewServerProofPool(len(serverProofs))
 53	for _, serverProof := range serverProofs {
 54		serverProofPool.Add(serverProof)
 55	}
 56	c := Coordinator{
 57		config:          conf,
 58		serverProofPool: serverProofPool,
 59		hdb:             hdb,
 60		txsel:           txsel,
 61		batchBuilder:    bb,
 62		ethClient:       ethClient,
 63		ethTxStore:      memory.NewMemoryStorage(),
 64	}
 65	return &c
 66}
 67
 68// ForgeLoopFn is the function ran in a loop that checks if it's time to forge
 69// and forges a batch if so and sends it to outBatchCh.  Returns true if it's
 70// the coordinator turn to forge.
 71func (c *Coordinator) ForgeLoopFn(outBatchCh chan *BatchInfo, stopCh chan bool) (bool, error) {
 72	if !c.isForgeSequence() {
 73		if c.forging {
 74			log.Info("stop forging")
 75			c.forging = false
 76		}
 77		log.Debug("not in forge time")
 78		return false, nil
 79	}
 80	log.Debug("forge time")
 81	if !c.forging {
 82		log.Info("start forging")
 83		// c.batchNum = c.hdb.GetLastBatchNum() // uncomment when HistoryDB is ready
 84		err := c.txsel.Reset(c.batchNum)
 85		if err != nil {
 86			log.Errorw("TxSelector.Reset", "error", err)
 87			return true, err
 88		}
 89		err = c.batchBuilder.Reset(c.batchNum, true)
 90		if err != nil {
 91			log.Errorw("BatchBuilder.Reset", "error", err)
 92			return true, err
 93		}
 94		// c.batchQueue = NewBatchQueue()
 95		c.forging = true
 96	}
 97	// TODO once synchronizer has this method ready:
 98	// If there's been a reorg, handle it
 99	// handleReorg() function decides if the reorg must restart the pipeline or not
100	// if c.synchronizer.Reorg():
101	_ = c.handleReorg()
102
103	// 0. If there's an available server proof: Start pipeline for batchNum = batchNum + 1.
104	// non-blocking call, returns nil if a server proof is
105	// not available, or non-nil otherwise.
106	serverProof, err := c.serverProofPool.Get(stopCh)
107	if err != nil {
108		return true, err
109	}
110	log.Debugw("got serverProof", "server", serverProof)
111
112	log.Debugw("start forge")
113	batchInfo, err := c.forge(serverProof)
114	if err != nil {
115		log.Errorw("forge", "error", err)
116		return true, err
117	}
118	log.Debugw("end forge", "batchNum", batchInfo.batchNum)
119	outBatchCh <- batchInfo
120	return true, nil
121}
122
123// GetProofCallForgeLoopFn is the function ran in a loop that gets a forged
124// batch via inBatchCh, waits for the proof server to finish, calls the ForgeBatch
125// function in the Rollup Smart Contract, and sends the batch to outBatchCh.
126func (c *Coordinator) GetProofCallForgeLoopFn(inBatchCh, outBatchCh chan *BatchInfo, stopCh chan bool) error {
127	select {
128	case <-stopCh:
129		log.Info("forgeLoopFn stopped")
130		return ErrStop
131	case batchInfo := <-inBatchCh:
132		log.Debugw("start getProofCallForge", "batchNum", batchInfo.batchNum)
133		if err := c.getProofCallForge(batchInfo, stopCh); err != nil {
134			return err
135		}
136		log.Debugw("end getProofCallForge", "batchNum", batchInfo.batchNum)
137		outBatchCh <- batchInfo
138	}
139	return nil
140}
141
142// ForgeCallConfirmLoopFn is the function ran in a loop that gets a batch that
143// has been sent to the Rollup Smart Contract via inBatchCh and waits for the
144// ethereum transaction confirmation.
145func (c *Coordinator) ForgeCallConfirmLoopFn(inBatchCh chan *BatchInfo, stopCh chan bool) error {
146	select {
147	case <-stopCh:
148		log.Info("forgeConfirmLoopFn stopped")
149		return ErrStop
150	case batchInfo := <-inBatchCh:
151		log.Debugw("start forgeCallConfirm", "batchNum", batchInfo.batchNum)
152		if err := c.forgeCallConfirm(batchInfo); err != nil {
153			return err
154		}
155		log.Debugw("end forgeCallConfirm", "batchNum", batchInfo.batchNum)
156	}
157	return nil
158}
159
160func (c *Coordinator) forge(serverProof ServerProofInterface) (*BatchInfo, error) {
161	// remove transactions from the pool that have been there for too long
162	err := c.purgeRemoveByTimeout()
163	if err != nil {
164		return nil, err
165	}
166
167	c.batchNum = c.batchNum + 1
168	batchInfo := NewBatchInfo(c.batchNum, serverProof) // to accumulate metadata of the batch
169
170	var poolL2Txs []*common.PoolL2Tx
171	// var feesInfo
172	var l1UserTxsExtra, l1OperatorTxs []*common.L1Tx
173	// 1. Decide if we forge L2Tx or L1+L2Tx
174	if c.shouldL1L2Batch() {
175		// 2a: L1+L2 txs
176		// l1UserTxs, toForgeL1TxsNumber := c.hdb.GetNextL1UserTxs() // TODO once HistoryDB is ready, uncomment
177		var l1UserTxs []*common.L1Tx = nil                                                                // tmp, depends on HistoryDB
178		l1UserTxsExtra, l1OperatorTxs, poolL2Txs, err = c.txsel.GetL1L2TxSelection(c.batchNum, l1UserTxs) // TODO once feesInfo is added to method return, add the var
179		if err != nil {
180			return nil, err
181		}
182	} else {
183		// 2b: only L2 txs
184		poolL2Txs, err = c.txsel.GetL2TxSelection(c.batchNum) // TODO once feesInfo is added to method return, add the var
185		if err != nil {
186			return nil, err
187		}
188		l1UserTxsExtra = nil
189		l1OperatorTxs = nil
190	}
191
192	// Run purger to invalidate transactions that become invalid beause of
193	// the poolL2Txs selected.  Will mark as invalid the txs that have a
194	// (fromIdx, nonce) which already appears in the selected txs (includes
195	// all the nonces smaller than the current one)
196	err = c.purgeInvalidDueToL2TxsSelection(poolL2Txs)
197	if err != nil {
198		return nil, err
199	}
200
201	// 3.  Save metadata from TxSelector output for BatchNum
202	batchInfo.SetTxsInfo(l1UserTxsExtra, l1OperatorTxs, poolL2Txs) // TODO feesInfo
203
204	// 4. Call BatchBuilder with TxSelector output
205	configBatch := &batchbuilder.ConfigBatch{
206		ForgerAddress: c.config.ForgerAddress,
207	}
208	zkInputs, err := c.batchBuilder.BuildBatch(configBatch, l1UserTxsExtra, l1OperatorTxs, poolL2Txs, nil) // TODO []common.TokenID --> feesInfo
209	if err != nil {
210		return nil, err
211	}
212
213	// 5. Save metadata from BatchBuilder output for BatchNum
214	batchInfo.SetZKInputs(zkInputs)
215
216	// 6. Call an idle server proof with BatchBuilder output, save server proof info for batchNum
217	err = batchInfo.serverProof.CalculateProof(zkInputs)
218	if err != nil {
219		return nil, err
220	}
221
222	return &batchInfo, nil
223}
224
225// getProofCallForge gets the generated zkProof & sends it to the SmartContract
226func (c *Coordinator) getProofCallForge(batchInfo *BatchInfo, stopCh chan bool) error {
227	serverProof := batchInfo.serverProof
228	proof, err := serverProof.GetProof(stopCh) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
229	if err != nil {
230		return err
231	}
232	batchInfo.SetProof(proof)
233	forgeBatchArgs := c.prepareForgeBatchArgs(batchInfo)
234	_, err = c.ethClient.RollupForgeBatch(forgeBatchArgs)
235	if err != nil {
236		return err
237	}
238	log.Debugf("ethClient ForgeCall sent, batchNum: %d", c.batchNum)
239
240	// TODO once tx data type is defined, store ethTx (returned by ForgeCall)
241	// TBD if use ethTxStore as a disk k-v database, or use a Queue
242	// tx, err := c.ethTxStore.NewTx()
243	// if err != nil {
244	//         return err
245	// }
246	// tx.Put(ethTx.Hash(), ethTx.Bytes())
247	// if err := tx.Commit(); err!=nil {
248	//         return nil
249	// }
250
251	return nil
252}
253
254func (c *Coordinator) forgeCallConfirm(batchInfo *BatchInfo) error {
255	// TODO strategy of this sequence TBD
256	// confirm eth txs and mark them as accepted sequence
257	// ethTx := ethTxStore.GetFirstPending()
258	// waitForAccepted(ethTx) // blocking call, returns once the ethTx is mined
259	// ethTxStore.MarkAccepted(ethTx)
260	return nil
261}
262
263func (c *Coordinator) handleReorg() error {
264	return nil // TODO
265}
266
267// isForgeSequence returns true if the node is the Forger in the current ethereum block
268func (c *Coordinator) isForgeSequence() bool {
269	return c.isForgeSeq // TODO
270}
271
272func (c *Coordinator) purgeRemoveByTimeout() error {
273	return nil // TODO
274}
275
276func (c *Coordinator) purgeInvalidDueToL2TxsSelection(l2Txs []*common.PoolL2Tx) error {
277	return nil // TODO
278}
279
280func (c *Coordinator) shouldL1L2Batch() bool {
281	return false // TODO
282}
283
284func (c *Coordinator) prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
285	return nil // TODO
286}