hermez-node/coordinator/coordinator.go
coordinator/coordinator.goRaw
1/*
2Package coordinator handles all the logic related to forging batches as a
3coordinator in the hermez network.
4
5The forging of batches is done with a pipeline in order to allow multiple
6batches being forged in parallel. The maximum number of batches that can be
7forged in parallel is determined by the number of available proof servers.
8
9The Coordinator begins with the pipeline stopped. The main Coordinator
10goroutine keeps listening for synchronizer events sent by the node package,
11which allow the coordinator to determine if the configured forger address is
12allowed to forge at the current block or not. When the forger address becomes
13allowed to forge, the pipeline is started, and when it terminates being allowed
14to forge, the pipeline is stopped.
15
16The Pipeline consists of two goroutines. The first one is in charge of
17preparing a batch internally, which involves making a selection of transactions
18and calculating the ZKInputs for the batch proof, and sending these ZKInputs to
19an idle proof server. This goroutine will keep preparing batches while there
20are idle proof servers, if the forging policy determines that a batch should be
21forged in the current state. The second goroutine is in charge of waiting for
22the proof server to finish computing the proof, retreiving it, prepare the
23arguments for the `forgeBatch` Rollup transaction, and sending the result to
24the TxManager. All the batch information moves between functions and
25goroutines via the BatchInfo struct.
26
27Finally, the TxManager contains a single goroutine that makes forgeBatch
28ethereum transactions for the batches sent by the Pipeline, and keeps them in a
29list to check them periodically. In the periodic checks, the ethereum
30transaction is checked for successfulness, and it's only forgotten after a
31number of confirmation blocks have passed after being successfully mined. At
32any point if a transaction failure is detected, the TxManager can signal the
33Coordinator to reset the Pipeline in order to reforge the failed batches.
34
35The Coordinator goroutine acts as a manager. The synchronizer events (which
36notify about new blocks and associated new state) that it receives are
37broadcasted to the Pipeline and the TxManager. This allows the Coordinator,
38Pipeline and TxManager to have a copy of the current hermez network state
39required to perform their duties.
40*/
41package coordinator
42
43import (
44 "context"
45 "fmt"
46 "math/big"
47 "os"
48 "sync"
49 "time"
50
51 ethCommon "github.com/ethereum/go-ethereum/common"
52 "github.com/hermeznetwork/hermez-node/batchbuilder"
53 "github.com/hermeznetwork/hermez-node/common"
54 "github.com/hermeznetwork/hermez-node/config"
55 "github.com/hermeznetwork/hermez-node/db/historydb"
56 "github.com/hermeznetwork/hermez-node/db/l2db"
57 "github.com/hermeznetwork/hermez-node/eth"
58 "github.com/hermeznetwork/hermez-node/log"
59 "github.com/hermeznetwork/hermez-node/prover"
60 "github.com/hermeznetwork/hermez-node/synchronizer"
61 "github.com/hermeznetwork/hermez-node/txprocessor"
62 "github.com/hermeznetwork/hermez-node/txselector"
63 "github.com/hermeznetwork/tracerr"
64)
65
66var (
67 errLastL1BatchNotSynced = fmt.Errorf("last L1Batch not synced yet")
68 errSkipBatchByPolicy = fmt.Errorf("skip batch by policy")
69)
70
71const (
72 queueLen = 16
73 longWaitDuration = 999 * time.Hour
74 zeroDuration = 0 * time.Second
75)
76
77// Config contains the Coordinator configuration
78type Config struct {
79 // ForgerAddress is the address under which this coordinator is forging
80 ForgerAddress ethCommon.Address
81 // ConfirmBlocks is the number of confirmation blocks to wait for sent
82 // ethereum transactions before forgetting about them
83 ConfirmBlocks int64
84 // L1BatchTimeoutPerc is the portion of the range before the L1Batch
85 // timeout that will trigger a schedule to forge an L1Batch
86 L1BatchTimeoutPerc float64
87 // StartSlotBlocksDelay is the number of blocks of delay to wait before
88 // starting the pipeline when we reach a slot in which we can forge.
89 StartSlotBlocksDelay int64
90 // ScheduleBatchBlocksAheadCheck is the number of blocks ahead in which
91 // the forger address is checked to be allowed to forge (apart from
92 // checking the next block), used to decide when to stop scheduling new
93 // batches (by stopping the pipeline).
94 // For example, if we are at block 10 and ScheduleBatchBlocksAheadCheck
95 // is 5, even though at block 11 we canForge, the pipeline will be
96 // stopped if we can't forge at block 15.
97 // This value should be the expected number of blocks it takes between
98 // scheduling a batch and having it mined.
99 ScheduleBatchBlocksAheadCheck int64
100 // SendBatchBlocksMarginCheck is the number of margin blocks ahead in
101 // which the coordinator is also checked to be allowed to forge, apart
102 // from the next block; used to decide when to stop sending batches to
103 // the smart contract.
104 // For example, if we are at block 10 and SendBatchBlocksMarginCheck is
105 // 5, even though at block 11 we canForge, the batch will be discarded
106 // if we can't forge at block 15.
107 // This value should be the expected number of blocks it takes between
108 // sending a batch and having it mined.
109 SendBatchBlocksMarginCheck int64
110 // EthClientAttempts is the number of attempts to do an eth client RPC
111 // call before giving up
112 EthClientAttempts int
113 // ForgeRetryInterval is the waiting interval between calls forge a
114 // batch after an error
115 ForgeRetryInterval time.Duration
116 // ForgeDelay is the delay after which a batch is forged if the slot is
117 // already committed. If set to 0s, the coordinator will continuously
118 // forge at the maximum rate.
119 ForgeDelay time.Duration
120 // ForgeNoTxsDelay is the delay after which a batch is forged even if
121 // there are no txs to forge if the slot is already committed. If set
122 // to 0s, the coordinator will continuously forge even if the batches
123 // are empty.
124 ForgeNoTxsDelay time.Duration
125 // MustForgeAtSlotDeadline enables the coordinator to forge slots if
126 // the empty slots reach the slot deadline.
127 MustForgeAtSlotDeadline bool
128 // IgnoreSlotCommitment disables forcing the coordinator to forge a
129 // slot immediately when the slot is not committed. If set to false,
130 // the coordinator will immediately forge a batch at the beginning of
131 // a slot if it's the slot winner.
132 IgnoreSlotCommitment bool
133 // ForgeOncePerSlotIfTxs will make the coordinator forge at most one
134 // batch per slot, only if there are included txs in that batch, or
135 // pending l1UserTxs in the smart contract. Setting this parameter
136 // overrides `ForgeDelay`, `ForgeNoTxsDelay`, `MustForgeAtSlotDeadline`
137 // and `IgnoreSlotCommitment`.
138 ForgeOncePerSlotIfTxs bool
139 // SyncRetryInterval is the waiting interval between calls to the main
140 // handler of a synced block after an error
141 SyncRetryInterval time.Duration
142 // PurgeByExtDelInterval is the waiting interval between calls
143 // to the PurgeByExternalDelete function of the l2db which deletes
144 // pending txs externally marked by the column `external_delete`
145 PurgeByExtDelInterval time.Duration
146 // EthClientAttemptsDelay is delay between attempts do do an eth client
147 // RPC call
148 EthClientAttemptsDelay time.Duration
149 // EthTxResendTimeout is the timeout after which a non-mined ethereum
150 // transaction will be resent (reusing the nonce) with a newly
151 // calculated gas price
152 EthTxResendTimeout time.Duration
153 // EthNoReuseNonce disables reusing nonces of pending transactions for
154 // new replacement transactions
155 EthNoReuseNonce bool
156 // MaxGasPrice is the maximum gas price allowed for ethereum
157 // transactions
158 MaxGasPrice *big.Int
159 // GasPriceIncPerc is the percentage increase of gas price set in an
160 // ethereum transaction from the suggested gas price by the ehtereum
161 // node
162 GasPriceIncPerc int64
163 // TxManagerCheckInterval is the waiting interval between receipt
164 // checks of ethereum transactions in the TxManager
165 TxManagerCheckInterval time.Duration
166 // DebugBatchPath if set, specifies the path where batchInfo is stored
167 // in JSON in every step/update of the pipeline
168 DebugBatchPath string
169 Purger PurgerCfg
170 // VerifierIdx is the index of the verifier contract registered in the
171 // smart contract
172 VerifierIdx uint8
173 // ForgeBatchGasCost contains the cost of each action in the
174 // ForgeBatch transaction.
175 ForgeBatchGasCost config.ForgeBatchGasCost
176 TxProcessorConfig txprocessor.Config
177}
178
179func (c *Config) debugBatchStore(batchInfo *BatchInfo) {
180 if c.DebugBatchPath != "" {
181 if err := batchInfo.DebugStore(c.DebugBatchPath); err != nil {
182 log.Warnw("Error storing debug BatchInfo",
183 "path", c.DebugBatchPath, "err", err)
184 }
185 }
186}
187
188type fromBatch struct {
189 BatchNum common.BatchNum
190 ForgerAddr ethCommon.Address
191 StateRoot *big.Int
192}
193
194// Coordinator implements the Coordinator type
195type Coordinator struct {
196 // State
197 pipelineNum int // Pipeline sequential number. The first pipeline is 1
198 pipelineFromBatch fromBatch // batch from which we started the pipeline
199 provers []prover.Client
200 consts common.SCConsts
201 vars common.SCVariables
202 stats synchronizer.Stats
203 started bool
204
205 cfg Config
206
207 historyDB *historydb.HistoryDB
208 l2DB *l2db.L2DB
209 txSelector *txselector.TxSelector
210 batchBuilder *batchbuilder.BatchBuilder
211
212 msgCh chan interface{}
213 ctx context.Context
214 wg sync.WaitGroup
215 cancel context.CancelFunc
216
217 // mutexL2DBUpdateDelete protects updates to the L2DB so that
218 // these two processes always happen exclusively:
219 // - Pipeline taking pending txs, running through the TxProcessor and
220 // marking selected txs as forging
221 // - Coordinator deleting pending txs that have been marked with
222 // `external_delete`.
223 // Without this mutex, the coordinator could delete a pending txs that
224 // has just been selected by the TxProcessor in the pipeline.
225 mutexL2DBUpdateDelete sync.Mutex
226 pipeline *Pipeline
227 lastNonFailedBatchNum common.BatchNum
228
229 purger *Purger
230 txManager *TxManager
231}
232
233// NewCoordinator creates a new Coordinator
234func NewCoordinator(cfg Config,
235 historyDB *historydb.HistoryDB,
236 l2DB *l2db.L2DB,
237 txSelector *txselector.TxSelector,
238 batchBuilder *batchbuilder.BatchBuilder,
239 serverProofs []prover.Client,
240 ethClient eth.ClientInterface,
241 scConsts *common.SCConsts,
242 initSCVars *common.SCVariables,
243) (*Coordinator, error) {
244 // nolint reason: hardcoded `1.0`, by design the percentage can't be over 100%
245 if cfg.L1BatchTimeoutPerc >= 1.0 { //nolint:gomnd
246 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.L1BatchTimeoutPerc (%v >= 1.0)",
247 cfg.L1BatchTimeoutPerc))
248 }
249 if cfg.EthClientAttempts < 1 {
250 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.EthClientAttempts (%v < 1)",
251 cfg.EthClientAttempts))
252 }
253
254 if cfg.DebugBatchPath != "" {
255 if err := os.MkdirAll(cfg.DebugBatchPath, 0744); err != nil {
256 return nil, tracerr.Wrap(err)
257 }
258 }
259
260 purger := Purger{
261 cfg: cfg.Purger,
262 lastPurgeBlock: 0,
263 lastPurgeBatch: 0,
264 lastInvalidateBlock: 0,
265 lastInvalidateBatch: 0,
266 }
267
268 ctx, cancel := context.WithCancel(context.Background())
269 c := Coordinator{
270 pipelineNum: 0,
271 pipelineFromBatch: fromBatch{
272 BatchNum: 0,
273 ForgerAddr: ethCommon.Address{},
274 StateRoot: big.NewInt(0),
275 },
276 provers: serverProofs,
277 consts: *scConsts,
278 vars: *initSCVars,
279
280 cfg: cfg,
281
282 historyDB: historyDB,
283 l2DB: l2DB,
284 txSelector: txSelector,
285 batchBuilder: batchBuilder,
286
287 purger: &purger,
288
289 msgCh: make(chan interface{}),
290 ctx: ctx,
291 // wg
292 cancel: cancel,
293 }
294 ctxTimeout, ctxTimeoutCancel := context.WithTimeout(ctx, 1*time.Second)
295 defer ctxTimeoutCancel()
296 txManager, err := NewTxManager(ctxTimeout, &cfg, ethClient, l2DB, &c,
297 scConsts, initSCVars)
298 if err != nil {
299 return nil, tracerr.Wrap(err)
300 }
301 c.txManager = txManager
302 // Set Eth LastBlockNum to -1 in stats so that stats.Synced() is
303 // guaranteed to return false before it's updated with a real stats
304 c.stats.Eth.LastBlock.Num = -1
305 return &c, nil
306}
307
308// TxSelector returns the inner TxSelector
309func (c *Coordinator) TxSelector() *txselector.TxSelector {
310 return c.txSelector
311}
312
313// BatchBuilder returns the inner BatchBuilder
314func (c *Coordinator) BatchBuilder() *batchbuilder.BatchBuilder {
315 return c.batchBuilder
316}
317
318func (c *Coordinator) newPipeline(ctx context.Context) (*Pipeline, error) {
319 c.pipelineNum++
320 return NewPipeline(ctx, c.cfg, c.pipelineNum, c.historyDB, c.l2DB, c.txSelector,
321 c.batchBuilder, &c.mutexL2DBUpdateDelete, c.purger, c, c.txManager,
322 c.provers, &c.consts)
323}
324
325// MsgSyncBlock indicates an update to the Synchronizer stats
326type MsgSyncBlock struct {
327 Stats synchronizer.Stats
328 Batches []common.BatchData
329 // Vars contains each Smart Contract variables if they are updated, or
330 // nil if they haven't changed.
331 Vars common.SCVariablesPtr
332}
333
334// MsgSyncReorg indicates a reorg
335type MsgSyncReorg struct {
336 Stats synchronizer.Stats
337 Vars common.SCVariablesPtr
338}
339
340// MsgStopPipeline indicates a signal to reset the pipeline
341type MsgStopPipeline struct {
342 Reason string
343 // FailedBatchNum indicates the first batchNum that failed in the
344 // pipeline. If FailedBatchNum is 0, it should be ignored.
345 FailedBatchNum common.BatchNum
346}
347
348// SendMsg is a thread safe method to pass a message to the Coordinator
349func (c *Coordinator) SendMsg(ctx context.Context, msg interface{}) {
350 select {
351 case c.msgCh <- msg:
352 case <-ctx.Done():
353 }
354}
355
356func updateSCVars(vars *common.SCVariables, update common.SCVariablesPtr) {
357 if update.Rollup != nil {
358 vars.Rollup = *update.Rollup
359 }
360 if update.Auction != nil {
361 vars.Auction = *update.Auction
362 }
363 if update.WDelayer != nil {
364 vars.WDelayer = *update.WDelayer
365 }
366}
367
368func (c *Coordinator) syncSCVars(vars common.SCVariablesPtr) {
369 updateSCVars(&c.vars, vars)
370}
371
372func canForge(auctionConstants *common.AuctionConstants, auctionVars *common.AuctionVariables,
373 currentSlot *common.Slot, nextSlot *common.Slot, addr ethCommon.Address, blockNum int64,
374 mustForgeAtDeadline bool) bool {
375 if blockNum < auctionConstants.GenesisBlockNum {
376 log.Infow("canForge: requested blockNum is < genesis", "blockNum", blockNum,
377 "genesis", auctionConstants.GenesisBlockNum)
378 return false
379 }
380 var slot *common.Slot
381 if currentSlot.StartBlock <= blockNum && blockNum <= currentSlot.EndBlock {
382 slot = currentSlot
383 } else if nextSlot.StartBlock <= blockNum && blockNum <= nextSlot.EndBlock {
384 slot = nextSlot
385 } else {
386 log.Warnw("canForge: requested blockNum is outside current and next slot",
387 "blockNum", blockNum, "currentSlot", currentSlot,
388 "nextSlot", nextSlot,
389 )
390 return false
391 }
392 anyoneForge := false
393 if !slot.ForgerCommitment &&
394 auctionConstants.RelativeBlock(blockNum) >= int64(auctionVars.SlotDeadline) {
395 log.Debugw("canForge: anyone can forge in the current slot (slotDeadline passed)",
396 "block", blockNum)
397 anyoneForge = true
398 }
399 if slot.Forger == addr || (anyoneForge && mustForgeAtDeadline) {
400 return true
401 }
402 log.Debugw("canForge: can't forge", "slot.Forger", slot.Forger)
403 return false
404}
405
406func (c *Coordinator) canForgeAt(blockNum int64) bool {
407 return canForge(&c.consts.Auction, &c.vars.Auction,
408 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
409 c.cfg.ForgerAddress, blockNum, c.cfg.MustForgeAtSlotDeadline)
410}
411
412func (c *Coordinator) canForge() bool {
413 blockNum := c.stats.Eth.LastBlock.Num + 1
414 return canForge(&c.consts.Auction, &c.vars.Auction,
415 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
416 c.cfg.ForgerAddress, blockNum, c.cfg.MustForgeAtSlotDeadline)
417}
418
419func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats) error {
420 nextBlock := c.stats.Eth.LastBlock.Num + 1
421 canForge := c.canForgeAt(nextBlock)
422 if c.cfg.ScheduleBatchBlocksAheadCheck != 0 && canForge {
423 canForge = c.canForgeAt(nextBlock + c.cfg.ScheduleBatchBlocksAheadCheck)
424 }
425 if c.pipeline == nil {
426 relativeBlock := c.consts.Auction.RelativeBlock(nextBlock)
427 if canForge && relativeBlock < c.cfg.StartSlotBlocksDelay {
428 log.Debugf("Coordinator: delaying pipeline start due to "+
429 "relativeBlock (%v) < cfg.StartSlotBlocksDelay (%v)",
430 relativeBlock, c.cfg.StartSlotBlocksDelay)
431 } else if canForge {
432 log.Infow("Coordinator: forging state begin", "block",
433 stats.Eth.LastBlock.Num+1, "batch", stats.Sync.LastBatch.BatchNum)
434 fromBatch := fromBatch{
435 BatchNum: stats.Sync.LastBatch.BatchNum,
436 ForgerAddr: stats.Sync.LastBatch.ForgerAddr,
437 StateRoot: stats.Sync.LastBatch.StateRoot,
438 }
439 if c.lastNonFailedBatchNum > fromBatch.BatchNum {
440 fromBatch.BatchNum = c.lastNonFailedBatchNum
441 fromBatch.ForgerAddr = c.cfg.ForgerAddress
442 fromBatch.StateRoot = big.NewInt(0)
443 }
444 // Before starting the pipeline make sure we reset any
445 // l2tx from the pool that was forged in a batch that
446 // didn't end up being mined. We are already doing
447 // this in handleStopPipeline, but we do it again as a
448 // failsafe in case the last synced batchnum is
449 // different than in the previous call to l2DB.Reorg,
450 // or in case the node was restarted when there was a
451 // started batch that included l2txs but was not mined.
452 if err := c.l2DB.Reorg(fromBatch.BatchNum); err != nil {
453 return tracerr.Wrap(err)
454 }
455 var err error
456 if c.pipeline, err = c.newPipeline(ctx); err != nil {
457 return tracerr.Wrap(err)
458 }
459 c.pipelineFromBatch = fromBatch
460 // Start the pipeline
461 if err := c.pipeline.Start(fromBatch.BatchNum, stats, &c.vars); err != nil {
462 c.pipeline = nil
463 return tracerr.Wrap(err)
464 }
465 }
466 } else {
467 if !canForge {
468 log.Infow("Coordinator: forging state end", "block", stats.Eth.LastBlock.Num+1)
469 c.pipeline.Stop(c.ctx)
470 c.pipeline = nil
471 }
472 }
473 if c.pipeline == nil {
474 if _, err := c.purger.InvalidateMaybe(c.l2DB, c.txSelector.LocalAccountsDB(),
475 stats.Sync.LastBlock.Num, int64(stats.Sync.LastBatch.BatchNum)); err != nil {
476 return tracerr.Wrap(err)
477 }
478 if _, err := c.purger.PurgeMaybe(c.l2DB, stats.Sync.LastBlock.Num,
479 int64(stats.Sync.LastBatch.BatchNum)); err != nil {
480 return tracerr.Wrap(err)
481 }
482 }
483 return nil
484}
485
486func (c *Coordinator) handleMsgSyncBlock(ctx context.Context, msg *MsgSyncBlock) error {
487 c.stats = msg.Stats
488 c.syncSCVars(msg.Vars)
489 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
490 if c.pipeline != nil {
491 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
492 }
493 if !c.stats.Synced() {
494 return nil
495 }
496 return c.syncStats(ctx, &c.stats)
497}
498
499func (c *Coordinator) handleReorg(ctx context.Context, msg *MsgSyncReorg) error {
500 c.stats = msg.Stats
501 c.syncSCVars(msg.Vars)
502 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
503 if c.pipeline != nil {
504 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
505 }
506 if c.stats.Sync.LastBatch.ForgerAddr != c.cfg.ForgerAddress &&
507 (c.stats.Sync.LastBatch.StateRoot == nil || c.pipelineFromBatch.StateRoot == nil ||
508 c.stats.Sync.LastBatch.StateRoot.Cmp(c.pipelineFromBatch.StateRoot) != 0) {
509 // There's been a reorg and the batch state root from which the
510 // pipeline was started has changed (probably because it was in
511 // a block that was discarded), and it was sent by a different
512 // coordinator than us. That batch may never be in the main
513 // chain, so we stop the pipeline (it will be started again
514 // once the node is in sync).
515 log.Infow("Coordinator.handleReorg StopPipeline sync.LastBatch.ForgerAddr != cfg.ForgerAddr "+
516 "& sync.LastBatch.StateRoot != pipelineFromBatch.StateRoot",
517 "sync.LastBatch.StateRoot", c.stats.Sync.LastBatch.StateRoot,
518 "pipelineFromBatch.StateRoot", c.pipelineFromBatch.StateRoot)
519 c.txManager.DiscardPipeline(ctx, c.pipelineNum)
520 if err := c.handleStopPipeline(ctx, "reorg", 0); err != nil {
521 return tracerr.Wrap(err)
522 }
523 }
524 return nil
525}
526
527// handleStopPipeline handles stopping the pipeline. If failedBatchNum is 0,
528// the next pipeline will start from the last state of the synchronizer,
529// otherwise, it will state from failedBatchNum-1.
530func (c *Coordinator) handleStopPipeline(ctx context.Context, reason string,
531 failedBatchNum common.BatchNum) error {
532 batchNum := c.stats.Sync.LastBatch.BatchNum
533 if failedBatchNum != 0 {
534 batchNum = failedBatchNum - 1
535 }
536 if c.pipeline != nil {
537 c.pipeline.Stop(c.ctx)
538 c.pipeline = nil
539 }
540 if err := c.l2DB.Reorg(batchNum); err != nil {
541 return tracerr.Wrap(err)
542 }
543 c.lastNonFailedBatchNum = batchNum
544 return nil
545}
546
547func (c *Coordinator) handleMsg(ctx context.Context, msg interface{}) error {
548 switch msg := msg.(type) {
549 case MsgSyncBlock:
550 if err := c.handleMsgSyncBlock(ctx, &msg); err != nil {
551 return tracerr.Wrap(fmt.Errorf("Coordinator.handleMsgSyncBlock error: %w", err))
552 }
553 case MsgSyncReorg:
554 if err := c.handleReorg(ctx, &msg); err != nil {
555 return tracerr.Wrap(fmt.Errorf("Coordinator.handleReorg error: %w", err))
556 }
557 case MsgStopPipeline:
558 log.Infow("Coordinator received MsgStopPipeline", "reason", msg.Reason)
559 if err := c.handleStopPipeline(ctx, msg.Reason, msg.FailedBatchNum); err != nil {
560 return tracerr.Wrap(fmt.Errorf("Coordinator.handleStopPipeline: %w", err))
561 }
562 default:
563 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
564 }
565 return nil
566}
567
568// Start the coordinator
569func (c *Coordinator) Start() {
570 if c.started {
571 log.Fatal("Coordinator already started")
572 }
573 c.started = true
574 c.wg.Add(1)
575 go func() {
576 c.txManager.Run(c.ctx)
577 c.wg.Done()
578 }()
579
580 c.wg.Add(1)
581 go func() {
582 timer := time.NewTimer(longWaitDuration)
583 for {
584 select {
585 case <-c.ctx.Done():
586 log.Info("Coordinator done")
587 c.wg.Done()
588 return
589 case msg := <-c.msgCh:
590 if err := c.handleMsg(c.ctx, msg); c.ctx.Err() != nil {
591 continue
592 } else if err != nil {
593 log.Errorw("Coordinator.handleMsg", "err", err)
594 if !timer.Stop() {
595 <-timer.C
596 }
597 timer.Reset(c.cfg.SyncRetryInterval)
598 continue
599 }
600 case <-timer.C:
601 timer.Reset(longWaitDuration)
602 if !c.stats.Synced() {
603 continue
604 }
605 if err := c.syncStats(c.ctx, &c.stats); c.ctx.Err() != nil {
606 continue
607 } else if err != nil {
608 log.Errorw("Coordinator.syncStats", "err", err)
609 if !timer.Stop() {
610 <-timer.C
611 }
612 timer.Reset(c.cfg.SyncRetryInterval)
613 continue
614 }
615 }
616 }
617 }()
618
619 c.wg.Add(1)
620 go func() {
621 for {
622 select {
623 case <-c.ctx.Done():
624 log.Info("Coordinator L2DB.PurgeByExternalDelete loop done")
625 c.wg.Done()
626 return
627 case <-time.After(c.cfg.PurgeByExtDelInterval):
628 c.mutexL2DBUpdateDelete.Lock()
629 if err := c.l2DB.PurgeByExternalDelete(); err != nil {
630 log.Errorw("L2DB.PurgeByExternalDelete", "err", err)
631 }
632 c.mutexL2DBUpdateDelete.Unlock()
633 }
634 }
635 }()
636}
637
638const stopCtxTimeout = 200 * time.Millisecond
639
640// Stop the coordinator
641func (c *Coordinator) Stop() {
642 if !c.started {
643 log.Fatal("Coordinator already stopped")
644 }
645 c.started = false
646 log.Infow("Stopping Coordinator...")
647 c.cancel()
648 c.wg.Wait()
649 if c.pipeline != nil {
650 ctx, cancel := context.WithTimeout(context.Background(), stopCtxTimeout)
651 defer cancel()
652 c.pipeline.Stop(ctx)
653 c.pipeline = nil
654 }
655}