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