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