Merge branch 'master' into split-resource-kind-storage-access

Author: MishkaRogachev

broadcastclients/broadcastclients.go

@@ -41,6 +41,7 @@ func (r *Router) AddBroadcastMessages(feedMessages []*message.BroadcastFeedMessa
 }
 
 type BroadcastClients struct {
+	stopwaiter.StopWaiter
 	primaryClients   []*broadcastclient.BroadcastClient
 	secondaryClients []*broadcastclient.BroadcastClient
 	secondaryURL     []string
@@ -135,17 +136,18 @@ func clearAndResetTicker(timer *time.Ticker, interval time.Duration) {
 }
 
 func (bcs *BroadcastClients) Start(ctx context.Context) {
-	bcs.primaryRouter.StopWaiter.Start(ctx, bcs.primaryRouter)
-	bcs.secondaryRouter.StopWaiter.Start(ctx, bcs.secondaryRouter)
+	bcs.StopWaiter.Start(ctx, bcs)
+	bcs.primaryRouter.StopWaiter.Start(bcs.GetContext(), bcs.primaryRouter)
+	bcs.secondaryRouter.StopWaiter.Start(bcs.GetContext(), bcs.secondaryRouter)
 
 	for _, client := range bcs.primaryClients {
-		client.Start(ctx)
+		client.Start(bcs.GetContext())
 	}
 
 	var lastConfirmed arbutil.MessageIndex
 	recentFeedItemsNew := make(map[arbutil.MessageIndex]time.Time, RECENT_FEED_INITIAL_MAP_SIZE)
 	recentFeedItemsOld := make(map[arbutil.MessageIndex]time.Time, RECENT_FEED_INITIAL_MAP_SIZE)
-	bcs.primaryRouter.LaunchThread(func(ctx context.Context) {
+	bcs.LaunchThread(func(ctx context.Context) {
 		recentFeedItemsCleanup := time.NewTicker(RECENT_FEED_ITEM_TTL)
 		startSecondaryFeedTimer := time.NewTicker(MAX_FEED_INACTIVE_TIME)
 		stopSecondaryFeedTimer := time.NewTicker(PRIMARY_FEED_UPTIME)
@@ -297,10 +299,13 @@ func (bcs *BroadcastClients) stopSecondaryFeed() {
 }
 
 func (bcs *BroadcastClients) StopAndWait() {
-	for _, client := range bcs.primaryClients {
+	for _, client := range bcs.secondaryClients {
 		client.StopAndWait()
 	}
-	for _, client := range bcs.secondaryClients {
+	for _, client := range bcs.primaryClients {
 		client.StopAndWait()
 	}
+	bcs.secondaryRouter.StopWaiter.StopAndWait()
+	bcs.primaryRouter.StopWaiter.StopAndWait()
+	bcs.StopWaiter.StopAndWait()
 }

changelog/pmikolajczyk-nit-4667.md

@@ -0,0 +1,4 @@
+### Fixed
+ - Fix ValidationSpawnerRetryWrapper lifecycle: reuse one wrapper per module root instead of creating and leaking one per validation
+ - Fix BroadcastClients launching coordination goroutine on child Router's StopWaiter instead of its own
+ - Fix ValidationServer and ExecutionSpawner missing StopAndWait for their children

staker/block_validator.go

@@ -75,13 +75,13 @@ func (t *WorkerThrottler) Release() {
 }
 
 type ThrottledValidationSpawner struct {
-	Spawner   validator.ValidationSpawner
+	Spawner   *retry_wrapper.ValidationSpawnerRetryWrapper
 	Throttler *WorkerThrottler
 }
 
 func NewThrottledValidationSpawner(spawner validator.ValidationSpawner) *ThrottledValidationSpawner {
 	return &ThrottledValidationSpawner{
-		Spawner:   spawner,
+		Spawner:   retry_wrapper.NewValidationSpawnerRetryWrapper(spawner),
 		Throttler: &WorkerThrottler{maxWorkers: spawner.Capacity()},
 	}
 }
@@ -996,9 +996,7 @@ func (v *BlockValidator) sendValidations(ctx context.Context) (*arbutil.MessageI
 		var runs []validator.ValidationRun
 		for _, moduleRoot := range wasmRoots {
 			throttledSpawner := v.chosenValidator[moduleRoot]
-			spawner := retry_wrapper.NewValidationSpawnerRetryWrapper(throttledSpawner.Spawner)
-			spawner.StopWaiter.Start(ctx, v)
-			input, err := validationStatus.Entry.ToInput(spawner.StylusArchs())
+			input, err := validationStatus.Entry.ToInput(throttledSpawner.Spawner.StylusArchs())
 			if err != nil && ctx.Err() == nil {
 				v.possiblyFatal(fmt.Errorf("%w: error preparing validation", err))
 				throttledSpawner.Throttler.Release()
@@ -1011,7 +1009,7 @@ func (v *BlockValidator) sendValidations(ctx context.Context) (*arbutil.MessageI
 				}
 				return nil, ctx.Err()
 			}
-			run := spawner.LaunchWithNAllowedAttempts(input, moduleRoot, v.config().ValidationSpawningAllowedAttempts, v.config().ValidationSpawningAllowedTimeouts)
+			run := throttledSpawner.Spawner.LaunchWithNAllowedAttempts(input, moduleRoot, v.config().ValidationSpawningAllowedAttempts, v.config().ValidationSpawningAllowedTimeouts)
 			log.Trace("sendValidations: launched", "pos", validationStatus.Entry.Pos, "moduleRoot", moduleRoot)
 			runs = append(runs, run)
 		}
@@ -1558,12 +1556,18 @@ func (v *BlockValidator) LaunchWorkthreadsWhenCaughtUp(ctx context.Context) {
 
 func (v *BlockValidator) Start(ctxIn context.Context) error {
 	v.StopWaiter.Start(ctxIn, v)
+	for _, throttled := range v.chosenValidator {
+		throttled.Spawner.Start(v.GetContext())
+	}
 	v.LaunchThread(v.LaunchWorkthreadsWhenCaughtUp)
 	v.CallIteratively(v.iterativeValidationPrint)
 	return nil
 }
 
 func (v *BlockValidator) StopAndWait() {
+	for _, throttled := range v.chosenValidator {
+		throttled.Spawner.StopAndWait()
+	}
 	v.StopWaiter.StopAndWait()
 }
 

validator/retry_wrapper/retry_wrapper.go

@@ -25,6 +25,10 @@ func NewValidationSpawnerRetryWrapper(spawner validator.ValidationSpawner) *Vali
 	}
 }
 
+func (v *ValidationSpawnerRetryWrapper) Start(ctx context.Context) {
+	v.StopWaiter.Start(ctx, v)
+}
+
 // LaunchWithNAllowedAttempts launches the validation with a specified number of
 // allowed attempts to retry in case of failure. Timeout errors have their own
 // separate counter (allowedTimeouts) since they typically represent transient

validator/valnode/redis/consumer.go

@@ -25,7 +25,8 @@ import (
 // RedisValidationClient producers.
 type ValidationServer struct {
 	stopwaiter.StopWaiter
-	spawner validator.ExecutionSpawner
+	spawner     validator.ExecutionSpawner
+	boldSpawner *ExecutionSpawner
 
 	// consumers stores moduleRoot to consumer mapping.
 	consumers map[common.Hash]*pubsub.Consumer[*validator.ValidationInput, validator.GoGlobalState]
@@ -59,7 +60,7 @@ func NewValidationServer(cfg *ValidationServerConfig, spawner validator.Executio
 
 func (s *ValidationServer) Start(ctx_in context.Context) {
 	s.StopWaiter.Start(ctx_in, s)
-	s.StartBoldSpawner(s.GetContext())
+	s.startBoldSpawner()
 	// Channel that all consumers use to indicate their readiness.
 	readyStreams := make(chan struct{}, len(s.consumers))
 	type workUnit struct {
@@ -192,12 +193,34 @@ func (s *ValidationServer) Start(ctx_in context.Context) {
 	}
 }
 
-func (s *ValidationServer) StartBoldSpawner(ctx context.Context) {
-	boldSpawner, err := NewExecutionSpawner(s.config, s.spawner)
+func (s *ValidationServer) StopOnly() {
+	if s.boldSpawner != nil {
+		s.boldSpawner.StopOnly()
+	}
+	for _, c := range s.consumers {
+		c.StopOnly()
+	}
+	s.StopWaiter.StopOnly()
+}
+
+func (s *ValidationServer) StopAndWait() {
+	if s.boldSpawner != nil {
+		s.boldSpawner.StopAndWait()
+	}
+	for _, c := range s.consumers {
+		c.StopAndWait()
+	}
+	s.StopWaiter.StopAndWait()
+}
+
+func (s *ValidationServer) startBoldSpawner() {
+	var err error
+	s.boldSpawner, err = NewExecutionSpawner(s.config, s.spawner)
 	if err != nil {
 		log.Error("creating redis execution spawner", "error", err)
+		return
 	}
-	boldSpawner.Start(ctx)
+	s.boldSpawner.Start(s.GetContext())
 }
 
 type ExecutionSpawner struct {
@@ -233,6 +256,13 @@ func NewExecutionSpawner(cfg *ValidationServerConfig, spawner validator.Executio
 	}, nil
 }
 
+func (s *ExecutionSpawner) StopAndWait() {
+	for _, c := range s.consumers {
+		c.StopAndWait()
+	}
+	s.StopWaiter.StopAndWait()
+}
+
 func (s *ExecutionSpawner) Start(ctx_in context.Context) {
 	s.StopWaiter.Start(ctx_in, s)
 	// Channel that all consumers use to indicate their readiness.