Merge pull request #4510 from OffchainLabs/pmikolajczyk/nit-4666-ordering-stop-wait

Fix `StopWaiter` lifecycle: ordering and context propagation

Author: pmikolajczyk41

arbnode/batch_poster.go

@@ -2071,9 +2071,9 @@ func (b *BatchPoster) GetBacklogEstimate() uint64 {
 }
 
 func (b *BatchPoster) Start(ctxIn context.Context) {
-	b.dataPoster.Start(ctxIn)
-	b.redisLock.Start(ctxIn)
 	b.StopWaiter.Start(ctxIn, b)
+	b.dataPoster.Start(b.GetContext())
+	b.redisLock.Start(b.GetContext())
 	b.LaunchThread(b.pollForReverts)
 	b.LaunchThread(b.pollForL1PriceData)
 	commonEphemeralErrorHandler := util.NewEphemeralErrorHandler(time.Minute, "", 0)
@@ -2157,9 +2157,9 @@ func (b *BatchPoster) Start(ctxIn context.Context) {
 }
 
 func (b *BatchPoster) StopAndWait() {
-	b.StopWaiter.StopAndWait()
-	b.dataPoster.StopAndWait()
 	b.redisLock.StopAndWait()
+	b.dataPoster.StopAndWait()
+	b.StopWaiter.StopAndWait()
 }
 
 type BoolRing struct {

changelog/pmikolajczyk-nit-4666.md

@@ -0,0 +1,2 @@
+### Fixed
+ - Fix StopWaiter lifecycle ordering: stop children before parent in StopAndWait, and pass managed context to children in Start

cmd/el-proxy/main.go

@@ -137,12 +137,12 @@ func NewExpressLaneProxy(
 
 func (p *ExpressLaneProxy) Start(ctx context.Context) {
 	p.StopWaiter.Start(ctx, p)
-	p.expressLaneTracker.Start(ctx)
+	p.expressLaneTracker.Start(p.GetContext())
 }
 
 func (p *ExpressLaneProxy) StopAndWait() {
-	p.StopWaiter.StopAndWait()
 	p.expressLaneTracker.StopAndWait()
+	p.StopWaiter.StopAndWait()
 }
 
 var ErrorInternalConnectionError = errors.New("internal connection error")

execution/gethexec/addressfilter/service.go

@@ -84,7 +84,7 @@ func (s *FilterService) Start(ctx context.Context) {
 		return s.config.PollInterval
 	})
 
-	s.addressChecker.Start(ctx)
+	s.addressChecker.Start(s.GetContext())
 
 	log.Info("address-filter service started",
 		"poll_interval", s.config.PollInterval,

execution/gethexec/express_lane_service.go

@@ -123,18 +123,20 @@ func (es *expressLaneService) Start(ctxIn context.Context) {
 	es.StopWaiter.Start(ctxIn, es)
 
 	if es.redisCoordinator != nil {
-		es.redisCoordinator.Start(ctxIn)
+		es.redisCoordinator.Start(es.GetContext())
 	}
 }
 
 func (es *expressLaneService) StopAndWait() {
-	es.StopWaiter.StopAndWait()
 	if es.redisCoordinator != nil {
 		es.redisCoordinator.StopAndWait()
 	}
+	// tracker is started by ExecutionNode, not by expressLaneService,
+	// but stopped here because no one else does it.
 	if es.tracker != nil {
 		es.tracker.StopAndWait()
 	}
+	es.StopWaiter.StopAndWait()
 }
 
 // DontCareSequence is a special sequence number that indicates a transaction should bypass the

execution/gethexec/sequencer.go

@@ -1864,13 +1864,15 @@ func (s TxSource) String() string {
 }
 
 func (s *Sequencer) StopAndWait() {
-	s.StopWaiter.StopAndWait()
-	if s.addressFilterService != nil {
-		s.addressFilterService.StopAndWait()
-	}
+	// expressLaneService is started by ExecutionNode via StartExpressLaneService,
+	// but stopped here because the Sequencer owns it.
 	if s.config().Timeboost.Enable && s.expressLaneService != nil {
 		s.expressLaneService.StopAndWait()
 	}
+	if s.addressFilterService != nil {
+		s.addressFilterService.StopAndWait()
+	}
+	s.StopWaiter.StopAndWait()
 	if s.txRetryQueue.Len() == 0 &&
 		len(s.txQueue) == 0 &&
 		s.nonceFailures.Len() == 0 &&

relay/relay.go

@@ -81,12 +81,12 @@ func (r *Relay) Start(ctx context.Context) error {
 	if err != nil {
 		return errors.New("broadcast unable to initialize")
 	}
-	err = r.broadcaster.Start(ctx)
+	err = r.broadcaster.Start(r.GetContext())
 	if err != nil {
 		return errors.New("broadcast unable to start")
 	}
 
-	r.broadcastClients.Start(ctx)
+	r.broadcastClients.Start(r.GetContext())
 
 	r.LaunchThread(func(ctx context.Context) {
 		for {
@@ -112,9 +112,9 @@ func (r *Relay) GetListenerAddr() net.Addr {
 }
 
 func (r *Relay) StopAndWait() {
-	r.StopWaiter.StopAndWait()
 	r.broadcastClients.StopAndWait()
 	r.broadcaster.StopAndWait()
+	r.StopWaiter.StopAndWait()
 }
 
 type Config struct {

staker/legacy/staker.go

@@ -561,10 +561,11 @@ func (s *Staker) getLatestStakedState(ctx context.Context, stakerAddress common.
 }
 
 func (s *Staker) StopAndWait() {
-	s.StopWaiter.StopAndWait()
+	// wallet is started externally, but stopped here because the Staker owns it.
 	if s.Strategy() != WatchtowerStrategy {
 		s.wallet.StopAndWait()
 	}
+	s.StopWaiter.StopAndWait()
 }
 
 func (s *Staker) Start(ctxIn context.Context) {

timeboost/auctioneer.go

@@ -445,15 +445,15 @@ func (a *AuctioneerServer) Start(ctx_in context.Context) {
 	a.StopWaiter.Start(ctx_in, a)
 	// Start S3 storage service to persist validated bids to s3
 	if a.s3StorageService != nil {
-		a.s3StorageService.Start(ctx_in)
+		a.s3StorageService.Start(a.GetContext())
 	}
 
 	// Start coordination to manage primary/secondary status
 	a.StopWaiter.CallIteratively(a.updateCoordination)
 
 	// Channel that consumer uses to indicate its readiness.
 	readyStream := make(chan struct{}, 1)
-	a.consumer.Start(ctx_in)
+	a.consumer.Start(a.GetContext())
 	// Channel for single consumer, once readiness is indicated in this,
 	// consumer will start consuming iteratively.
 	ready := make(chan struct{}, 1)
@@ -795,6 +795,9 @@ func (a *AuctioneerServer) StopAndWait() {
 	// auctioneerLivenessTimeout. This timeout gives time for existing messages to become
 	// unclaimed after IdleTimeToAutoclaim before the secondary auctioneer starts consuming
 	// messages.
-	a.StopWaiter.StopAndWait()
 	a.consumer.StopAndWait()
+	if a.s3StorageService != nil {
+		a.s3StorageService.StopAndWait()
+	}
+	a.StopWaiter.StopAndWait()
 }

timeboost/bid_validator.go

@@ -226,7 +226,7 @@ func (bv *BidValidator) Start(ctx_in context.Context) {
 	if bv.producer == nil {
 		log.Crit("Bid validator not yet initialized by calling Initialize(ctx)")
 	}
-	bv.producer.Start(ctx_in)
+	bv.producer.Start(bv.GetContext())
 
 	// Thread to set reserve price and clear per-round map of bid count per account.
 	bv.StopWaiter.LaunchThread(func(ctx context.Context) {