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