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