You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

311 lines
8.6 KiB

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