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