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