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