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