Merge pull request #4536 from OffchainLabs/pmikolajczyk/nit-4684-stop-waiter-api

Automatic child lifecycle management in StopWaiter

Author: pmikolajczyk41

arbnode/batch_poster.go

@@ -2073,8 +2073,8 @@ func (b *BatchPoster) GetBacklogEstimate() uint64 {
 
 func (b *BatchPoster) Start(ctxIn context.Context) {
 	b.StopWaiter.Start(ctxIn, b)
-	b.dataPoster.Start(b.GetContext())
-	b.redisLock.Start(b.GetContext())
+	b.StartAndTrackChild(b.dataPoster)
+	b.StartAndTrackChild(b.redisLock)
 	b.LaunchThread(b.pollForReverts)
 	b.LaunchThread(b.pollForL1PriceData)
 	commonEphemeralErrorHandler := util.NewEphemeralErrorHandler(time.Minute, "", 0)
@@ -2157,12 +2157,6 @@ func (b *BatchPoster) Start(ctxIn context.Context) {
 	})
 }
 
-func (b *BatchPoster) StopAndWait() {
-	b.redisLock.StopAndWait()
-	b.dataPoster.StopAndWait()
-	b.StopWaiter.StopAndWait()
-}
-
 type BoolRing struct {
 	buffer         []bool
 	bufferPosition int

arbnode/inbox_test.go

@@ -115,8 +115,7 @@ func (w *execClientWrapper) ArbOSVersionForMessageIndex(msgIdx arbutil.MessageIn
 	return w.ExecutionEngine.ArbOSVersionForMessageIndex(msgIdx)
 }
 
-func (w *execClientWrapper) StopAndWait() {
-}
+func (w *execClientWrapper) StopAndWait() {}
 
 func NewTransactionStreamerForTest(t *testing.T, ctx context.Context, ownerAddress common.Address) (*gethexec.ExecutionEngine, *TransactionStreamer, ethdb.Database, *core.BlockChain) {
 	chainConfig := chaininfo.ArbitrumDevTestChainConfig()

blocks_reexecutor/blocks_reexecutor.go

@@ -349,10 +349,6 @@ func (s *BlocksReExecutor) wrapFatalErr(err error) error {
 	return fmt.Errorf("shutting BlocksReExecutor down due to fatal error: %w", err)
 }
 
-func (s *BlocksReExecutor) StopAndWait() {
-	s.StopWaiter.StopAndWait()
-}
-
 func (s *BlocksReExecutor) dereferenceRoot(root common.Hash) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()

bold/challenge/manager.go

@@ -42,8 +42,7 @@ type Opt = func(val *Manager)
 // AssertionManager works with the challenge manager suppplying information
 // about assertions.
 type AssertionManager interface {
-	Start(context.Context)
-	StopAndWait()
+	stopwaiter.StoppableChild
 	LatestAgreedAssertion() protocol.AssertionHash
 	SetRivalHandler(types.RivalHandler)
 }
@@ -280,8 +279,7 @@ func (m *Manager) Start(ctx context.Context) {
 	m.StopWaiter.Start(ctx, m)
 	log.Info("Started challenge manager", "stakerAddress", m.chain.StakerAddress().Hex())
 
-	// Start the assertion manager on its own StopWaiter.
-	m.assertionManager.Start(m.GetContext())
+	m.StartAndTrackChild(m.assertionManager)
 
 	// Watcher tower and resolve modes don't monitor challenges.
 	if m.mode == types.WatchTowerMode || m.mode == types.ResolveMode {
@@ -291,26 +289,13 @@ func (m *Manager) Start(ctx context.Context) {
 	// Start watching for parent chain block events in the background.
 	m.LaunchThread(m.listenForBlockEvents)
 
-	// Start watching for ongoing chain events on its own StopWaiter.
-	m.watcher.Start(m.GetContext())
+	m.StartAndTrackChild(m.watcher)
 
 	if m.api != nil {
-		// Start the API server on its own StopWaiter.
-		m.api.Start(m.GetContext())
+		m.StartAndTrackChild(m.api)
 	}
 }
 
-func (m *Manager) StopAndWait() {
-	// Stop children first so they can shut down gracefully before
-	// the parent context is cancelled.
-	if m.api != nil {
-		m.api.StopAndWait()
-	}
-	m.watcher.StopAndWait()
-	m.assertionManager.StopAndWait()
-	m.StopWaiter.StopAndWait()
-}
-
 func (m *Manager) listenForBlockEvents(ctx context.Context) {
 	// If the chain watcher has not yet scraped and caught up all BoLD
 	// events up to the latest head, then we fire "block notification" events

bold/testing/e2e/headers.go

@@ -42,10 +42,6 @@ func (s *simpleHeaderProvider) listenToHeaders(ctx context.Context) {
 	}
 }
 
-func (s *simpleHeaderProvider) StopAndWait() {
-	s.StopWaiter.StopAndWait()
-}
-
 func (s *simpleHeaderProvider) Subscribe(requireBlockNrUpdates bool) (<-chan *types.Header, func()) {
 	ch := make(chan *types.Header, 100)
 	s.chs = append(s.chs, ch)

broadcastclients/broadcastclients.go

@@ -138,10 +138,13 @@ func clearAndResetTicker(timer *time.Ticker, interval time.Duration) {
 func (bcs *BroadcastClients) Start(ctx context.Context) {
 	bcs.StopWaiter.Start(ctx, bcs)
 	bcs.primaryRouter.StopWaiter.Start(bcs.GetContext(), bcs.primaryRouter)
+	bcs.TrackChild(bcs.primaryRouter)
 	bcs.secondaryRouter.StopWaiter.Start(bcs.GetContext(), bcs.secondaryRouter)
+	bcs.TrackChild(bcs.secondaryRouter)
 
 	for _, client := range bcs.primaryClients {
 		client.Start(bcs.GetContext())
+		bcs.TrackChild(client)
 	}
 
 	var lastConfirmed arbutil.MessageIndex
@@ -272,6 +275,8 @@ func (bcs *BroadcastClients) startSecondaryFeed(ctx context.Context) {
 		}
 		bcs.secondaryClients = append(bcs.secondaryClients, client)
 		client.Start(ctx)
+		// Secondary clients are not tracked — they are managed
+		// manually by startSecondaryFeed/stopSecondaryFeed.
 		log.Info("secondary feed started", "url", url, "startingFromSeq", latestSeqNum)
 	} else if len(bcs.secondaryURL) > 0 {
 		log.Warn("failed to start a new secondary feed all available secondary feeds were started")
@@ -299,13 +304,10 @@ func (bcs *BroadcastClients) stopSecondaryFeed() {
 }
 
 func (bcs *BroadcastClients) StopAndWait() {
+	// Secondary clients are not tracked (they're managed dynamically),
+	// so stop them explicitly for clean websocket shutdown.
 	for _, client := range bcs.secondaryClients {
 		client.StopAndWait()
 	}
-	for _, client := range bcs.primaryClients {
-		client.StopAndWait()
-	}
-	bcs.secondaryRouter.StopWaiter.StopAndWait()
-	bcs.primaryRouter.StopWaiter.StopAndWait()
 	bcs.StopWaiter.StopAndWait()
 }

changelog/pmikolajczyk-nit-4684.md

@@ -0,0 +1,2 @@
+### Internal
+- Add TrackChild/StartAndTrackChild to StopWaiter for automatic LIFO child lifecycle management

cmd/el-proxy/main.go

@@ -136,12 +136,7 @@ func NewExpressLaneProxy(
 
 func (p *ExpressLaneProxy) Start(ctx context.Context) {
 	p.StopWaiter.Start(ctx, p)
-	p.expressLaneTracker.Start(p.GetContext())
-}
-
-func (p *ExpressLaneProxy) StopAndWait() {
-	p.expressLaneTracker.StopAndWait()
-	p.StopWaiter.StopAndWait()
+	p.StartAndTrackChild(p.expressLaneTracker)
 }
 
 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(s.GetContext())
+	s.StartAndTrackChild(s.addressChecker)
 
 	log.Info("address-filter service started",
 		"poll_interval", s.config.PollInterval,

execution/gethexec/blockmetadata.go

@@ -111,7 +111,3 @@ func (b *BulkBlockMetadataFetcher) Start(ctx context.Context) {
 		_ = stopwaiter.CallWhenTriggeredWith[struct{}](&b.StopWaiterSafe, b.ClearCache, b.reorgDetector)
 	}
 }
-
-func (b *BulkBlockMetadataFetcher) StopAndWait() {
-	b.StopWaiter.StopAndWait()
-}