1package coordinator
  2
  3import (
  4	"time"
  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// CoordinatorConfig contains the Coordinator configuration
 18type CoordinatorConfig struct {
 19	ForgerAddress ethCommon.Address
 20	LoopInterval  time.Duration
 21}
 22
 23// Coordinator implements the Coordinator type
 24type Coordinator struct {
 25	// m            sync.Mutex
 26	stopch       chan bool
 27	stopforgerch chan bool
 28
 29	forging    bool
 30	isForgeSeq bool // WIP just for testing while implementing
 31
 32	config CoordinatorConfig
 33
 34	batchNum        common.BatchNum
 35	batchQueue      *BatchQueue
 36	serverProofPool ServerProofPool
 37
 38	// synchronizer *synchronizer.Synchronizer
 39	hdb          *historydb.HistoryDB
 40	txsel        *txselector.TxSelector
 41	batchBuilder *batchbuilder.BatchBuilder
 42
 43	ethClient  *eth.Client
 44	ethTxStore kvdb.Storage
 45}
 46
 47// NewCoordinator creates a new Coordinator
 48func NewCoordinator(conf CoordinatorConfig,
 49	hdb *historydb.HistoryDB,
 50	txsel *txselector.TxSelector,
 51	bb *batchbuilder.BatchBuilder,
 52	ethClient *eth.Client) *Coordinator { // once synchronizer is ready, synchronizer.Synchronizer will be passed as parameter here
 53	c := Coordinator{
 54		config:       conf,
 55		hdb:          hdb,
 56		txsel:        txsel,
 57		batchBuilder: bb,
 58		ethClient:    ethClient,
 59		ethTxStore:   memory.NewMemoryStorage(),
 60	}
 61	return &c
 62}
 63
 64func (c *Coordinator) Stop() {
 65	log.Info("Stopping Coordinator")
 66	c.stopch <- true
 67}
 68
 69// Start starts the Coordinator service
 70func (c *Coordinator) Start() {
 71	c.stopch = make(chan bool) // initialize channel
 72	go func() {
 73		log.Info("Starting Coordinator")
 74		for {
 75			select {
 76			case <-c.stopch:
 77				close(c.stopforgerch)
 78				log.Info("Coordinator stopped")
 79				return
 80			case <-time.After(c.config.LoopInterval):
 81				if !c.isForgeSequence() {
 82					if c.forging {
 83						log.Info("forging stopped")
 84						c.forging = false
 85						close(c.stopforgerch)
 86					}
 87					log.Debug("not in forge time")
 88					continue
 89				}
 90				if !c.forging {
 91					log.Info("Start Forging")
 92					// c.batchNum = c.hdb.GetLastBatchNum() // uncomment when HistoryDB is ready
 93					err := c.txsel.Reset(c.batchNum)
 94					if err != nil {
 95						log.Error("forging err: ", err)
 96					}
 97					err = c.batchBuilder.Reset(c.batchNum, true)
 98					if err != nil {
 99						log.Error("forging err: ", err)
100					}
101					c.batchQueue = NewBatchQueue()
102
103					c.forgerLoop()
104					c.forging = true
105				}
106			}
107		}
108	}()
109}
110
111// forgerLoop trigers goroutines for:
112// - forgeSequence
113// - proveSequence
114// - forgeConfirmationSequence
115func (c *Coordinator) forgerLoop() {
116	c.stopforgerch = make(chan bool) // initialize channel
117
118	go func() {
119		log.Info("forgeSequence started")
120		for {
121			select {
122			case <-c.stopforgerch:
123				log.Info("forgeSequence stopped")
124				return
125			case <-time.After(c.config.LoopInterval):
126				if err := c.forgeSequence(); err != nil {
127					log.Error("forgeSequence err: ", err)
128				}
129			}
130		}
131	}()
132	go func() {
133		log.Info("proveSequence started")
134		for {
135			select {
136			case <-c.stopforgerch:
137				log.Info("proveSequence stopped")
138				return
139			case <-time.After(c.config.LoopInterval):
140				if err := c.proveSequence(); err != nil && err != common.ErrBatchQueueEmpty {
141					log.Error("proveSequence err: ", err)
142				}
143			}
144		}
145	}()
146	go func() {
147		log.Info("forgeConfirmationSequence started")
148		for {
149			select {
150			case <-c.stopforgerch:
151				log.Info("forgeConfirmationSequence stopped")
152				return
153			case <-time.After(c.config.LoopInterval):
154				if err := c.forgeConfirmationSequence(); err != nil {
155					log.Error("forgeConfirmationSequence err: ", err)
156				}
157			}
158		}
159	}()
160}
161
162// forgeSequence
163func (c *Coordinator) forgeSequence() error {
164	// TODO once synchronizer has this method ready:
165	// If there's been a reorg, handle it
166	// handleReorg() function decides if the reorg must restart the pipeline or not
167	// if c.synchronizer.Reorg():
168	_ = c.handleReorg()
169
170	// 0. If there's an available server proof: Start pipeline for batchNum = batchNum + 1
171	serverProofInfo, err := c.serverProofPool.GetNextAvailable() // blocking call, returns when a server proof is available
172	if err != nil {
173		return err
174	}
175
176	// remove transactions from the pool that have been there for too long
177	err = c.purgeRemoveByTimeout()
178	if err != nil {
179		return err
180	}
181
182	c.batchNum = c.batchNum + 1
183	batchInfo := NewBatchInfo(c.batchNum, serverProofInfo) // to accumulate metadata of the batch
184
185	var poolL2Txs []*common.PoolL2Tx
186	// var feesInfo
187	var l1UserTxsExtra, l1OperatorTxs []*common.L1Tx
188	// 1. Decide if we forge L2Tx or L1+L2Tx
189	if c.shouldL1L2Batch() {
190		// 2a: L1+L2 txs
191		// l1UserTxs, toForgeL1TxsNumber := c.hdb.GetNextL1UserTxs() // TODO once HistoryDB is ready, uncomment
192		var l1UserTxs []*common.L1Tx = nil                                                                // tmp, depends on HistoryDB
193		l1UserTxsExtra, l1OperatorTxs, poolL2Txs, err = c.txsel.GetL1L2TxSelection(c.batchNum, l1UserTxs) // TODO once feesInfo is added to method return, add the var
194		if err != nil {
195			return err
196		}
197	} else {
198		// 2b: only L2 txs
199		poolL2Txs, err = c.txsel.GetL2TxSelection(c.batchNum) // TODO once feesInfo is added to method return, add the var
200		if err != nil {
201			return err
202		}
203		l1UserTxsExtra = nil
204		l1OperatorTxs = nil
205	}
206
207	// Run purger to invalidate transactions that become invalid beause of
208	// the poolL2Txs selected.  Will mark as invalid the txs that have a
209	// (fromIdx, nonce) which already appears in the selected txs (includes
210	// all the nonces smaller than the current one)
211	err = c.purgeInvalidDueToL2TxsSelection(poolL2Txs)
212	if err != nil {
213		return err
214	}
215
216	// 3.  Save metadata from TxSelector output for BatchNum
217	batchInfo.SetTxsInfo(l1UserTxsExtra, l1OperatorTxs, poolL2Txs) // TODO feesInfo
218
219	// 4. Call BatchBuilder with TxSelector output
220	configBatch := &batchbuilder.ConfigBatch{
221		ForgerAddress: c.config.ForgerAddress,
222	}
223	l2Txs := common.PoolL2TxsToL2Txs(poolL2Txs)
224	zkInputs, err := c.batchBuilder.BuildBatch(configBatch, l1UserTxsExtra, l1OperatorTxs, l2Txs, nil) // TODO []common.TokenID --> feesInfo
225	if err != nil {
226		return err
227	}
228
229	// 5. Save metadata from BatchBuilder output for BatchNum
230	batchInfo.SetZKInputs(zkInputs)
231	log.Debugf("Batch builded, batchNum: %d ", c.batchNum)
232
233	// 6. Call an idle server proof with BatchBuilder output, save server proof info for batchNum
234	err = batchInfo.serverProof.CalculateProof(zkInputs)
235	if err != nil {
236		return err
237	}
238	c.batchQueue.Push(&batchInfo)
239
240	return nil
241}
242
243// proveSequence gets the generated zkProof & sends it to the SmartContract
244func (c *Coordinator) proveSequence() error {
245	batchInfo := c.batchQueue.Pop()
246	if batchInfo == nil {
247		// no batches in queue, return
248		log.Debug("not batch to prove yet")
249		return common.ErrBatchQueueEmpty
250	}
251	serverProofInfo := batchInfo.serverProof
252	proof, err := serverProofInfo.GetProof() // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
253	if err != nil {
254		return err
255	}
256	batchInfo.SetProof(proof)
257	callData := c.prepareCallDataForge(batchInfo)
258	_, err = c.ethClient.ForgeCall(callData)
259	if err != nil {
260		return err
261	}
262	log.Debugf("ethClient ForgeCall sent, batchNum: %d", c.batchNum)
263
264	// TODO once tx data type is defined, store ethTx (returned by ForgeCall)
265	// TBD if use ethTxStore as a disk k-v database, or use a Queue
266	// tx, err := c.ethTxStore.NewTx()
267	// if err != nil {
268	//         return err
269	// }
270	// tx.Put(ethTx.Hash(), ethTx.Bytes())
271	// if err := tx.Commit(); err!=nil {
272	//         return nil
273	// }
274
275	return nil
276}
277
278func (c *Coordinator) forgeConfirmationSequence() error {
279	// TODO strategy of this sequence TBD
280	// confirm eth txs and mark them as accepted sequence
281	// ethTx := ethTxStore.GetFirstPending()
282	// waitForAccepted(ethTx) // blocking call, returns once the ethTx is mined
283	// ethTxStore.MarkAccepted(ethTx)
284	return nil
285}
286
287func (c *Coordinator) handleReorg() error {
288	return nil
289}
290
291// isForgeSequence returns true if the node is the Forger in the current ethereum block
292func (c *Coordinator) isForgeSequence() bool {
293	return c.isForgeSeq
294}
295
296func (c *Coordinator) purgeRemoveByTimeout() error {
297	return nil
298}
299
300func (c *Coordinator) purgeInvalidDueToL2TxsSelection(l2Txs []*common.PoolL2Tx) error {
301	return nil
302}
303
304func (c *Coordinator) shouldL1L2Batch() bool {
305	return false
306}
307
308func (c *Coordinator) prepareCallDataForge(batchInfo *BatchInfo) *common.CallDataForge {
309	return nil
310}