Integrate Redis streams in to Nitro's validation

pubsub/producer.go

@@ -78,6 +78,7 @@ var TestProducerConfig = ProducerConfig{
 func ProducerAddConfigAddOptions(prefix string, f *pflag.FlagSet) {
 	f.Bool(prefix+".enable-reproduce", DefaultProducerConfig.EnableReproduce, "when enabled, messages with dead consumer will be re-inserted into the stream")
 	f.Duration(prefix+".check-pending-interval", DefaultProducerConfig.CheckPendingInterval, "interval in which producer checks pending messages whether consumer processing them is inactive")
+	f.Duration(prefix+".check-result-interval", DefaultProducerConfig.CheckResultInterval, "interval in which producer checks pending messages whether consumer processing them is inactive")
 	f.Duration(prefix+".keepalive-timeout", DefaultProducerConfig.KeepAliveTimeout, "timeout after which consumer is considered inactive if heartbeat wasn't performed")
 }
 

staker/block_validator.go

@@ -146,6 +146,8 @@ type BlockValidatorConfigFetcher func() *BlockValidatorConfig
 func BlockValidatorConfigAddOptions(prefix string, f *pflag.FlagSet) {
 	f.Bool(prefix+".enable", DefaultBlockValidatorConfig.Enable, "enable block-by-block validation")
 	rpcclient.RPCClientAddOptions(prefix+".validation-server", f, &DefaultBlockValidatorConfig.ValidationServer)
+	rpcclient.RPCClientAddOptions(prefix+".execution-server-config", f, &DefaultBlockValidatorConfig.ExecutionServerConfig)
+	validatorclient.RedisValidationClientConfigAddOptions(prefix+".redis-validation-client-config", f)
 	f.String(prefix+".validation-server-configs-list", DefaultBlockValidatorConfig.ValidationServerConfigsList, "array of validation rpc configs given as a json string. time duration should be supplied in number indicating nanoseconds")
 	f.Duration(prefix+".validation-poll", DefaultBlockValidatorConfig.ValidationPoll, "poll time to check validations")
 	f.Uint64(prefix+".forward-blocks", DefaultBlockValidatorConfig.ForwardBlocks, "prepare entries for up to that many blocks ahead of validation (small footprint)")
@@ -165,6 +167,8 @@ var DefaultBlockValidatorConfig = BlockValidatorConfig{
 	Enable:                      false,
 	ValidationServerConfigsList: "default",
 	ValidationServer:            rpcclient.DefaultClientConfig,
+	ExecutionServerConfig:       rpcclient.DefaultClientConfig,
+	RedisValidationClientConfig: validatorclient.DefaultRedisValidationClientConfig,
 	ValidationPoll:              time.Second,
 	ForwardBlocks:               1024,
 	PrerecordedBlocks:           uint64(2 * runtime.NumCPU()),
@@ -176,18 +180,19 @@ var DefaultBlockValidatorConfig = BlockValidatorConfig{
 }
 
 var TestBlockValidatorConfig = BlockValidatorConfig{
-	Enable:                   false,
-	ValidationServer:         rpcclient.TestClientConfig,
-	ValidationServerConfigs:  []rpcclient.ClientConfig{rpcclient.TestClientConfig},
-	ExecutionServerConfig:    rpcclient.TestClientConfig,
-	ValidationPoll:           100 * time.Millisecond,
-	ForwardBlocks:            128,
-	PrerecordedBlocks:        uint64(2 * runtime.NumCPU()),
-	CurrentModuleRoot:        "latest",
-	PendingUpgradeModuleRoot: "latest",
-	FailureIsFatal:           true,
-	Dangerous:                DefaultBlockValidatorDangerousConfig,
-	MemoryFreeLimit:          "default",
+	Enable:                      false,
+	ValidationServer:            rpcclient.TestClientConfig,
+	ValidationServerConfigs:     []rpcclient.ClientConfig{rpcclient.TestClientConfig},
+	RedisValidationClientConfig: validatorclient.TestRedisValidationClientConfig,
+	ExecutionServerConfig:       rpcclient.TestClientConfig,
+	ValidationPoll:              100 * time.Millisecond,
+	ForwardBlocks:               128,
+	PrerecordedBlocks:           uint64(2 * runtime.NumCPU()),
+	CurrentModuleRoot:           "latest",
+	PendingUpgradeModuleRoot:    "latest",
+	FailureIsFatal:              true,
+	Dangerous:                   DefaultBlockValidatorDangerousConfig,
+	MemoryFreeLimit:             "default",
 }
 
 var DefaultBlockValidatorDangerousConfig = BlockValidatorDangerousConfig{
@@ -1060,6 +1065,9 @@ func (v *BlockValidator) Initialize(ctx context.Context) error {
 		}
 	}
 	log.Info("BlockValidator initialized", "current", v.currentWasmModuleRoot, "pending", v.pendingWasmModuleRoot)
+	if err := v.StatelessBlockValidator.Initialize([]common.Hash{v.currentWasmModuleRoot, v.pendingWasmModuleRoot}); err != nil {
+		return fmt.Errorf("initializing block validator with module roots: %w", err)
+	}
 	return nil
 }
 

staker/stateless_block_validator.go

@@ -199,7 +199,6 @@ func NewStatelessBlockValidator(
 		redisValClient, err := validatorclient.NewRedisValidationClient(&config().RedisValidationClientConfig)
 		if err != nil {
 			return nil, fmt.Errorf("creating new redis validation client: %w", err)
-			// log.Error("Creating redis validation client, redis validator disabled", "error", err)
 		}
 		validationSpawners = append(validationSpawners, redisValClient)
 	}
@@ -208,7 +207,10 @@ func NewStatelessBlockValidator(
 		validationSpawners = append(validationSpawners, validatorclient.NewValidationClient(valConfFetcher, stack))
 	}
 
-	validator := &StatelessBlockValidator{
+	valConfFetcher := func() *rpcclient.ClientConfig {
+		return &config().ExecutionServerConfig
+	}
+	return &StatelessBlockValidator{
 		config:             config(),
 		recorder:           recorder,
 		validationSpawners: validationSpawners,
@@ -218,12 +220,21 @@ func NewStatelessBlockValidator(
 		db:                 arbdb,
 		daService:          das,
 		blobReader:         blobReader,
+		execSpawner:        validatorclient.NewExecutionClient(valConfFetcher, stack),
+	}, nil
+}
+
+func (v *StatelessBlockValidator) Initialize(moduleRoots []common.Hash) error {
+	if len(v.validationSpawners) == 0 {
+		return nil
 	}
-	valConfFetcher := func() *rpcclient.ClientConfig {
-		return &config().ExecutionServerConfig
+	// First spawner is always RedisValidationClient if RedisStreams are enabled.
+	if v, ok := v.validationSpawners[0].(*validatorclient.RedisValidationClient); ok {
+		if err := v.Initialize(moduleRoots); err != nil {
+			return fmt.Errorf("initializing redis validation client module roots: %w", err)
+		}
 	}
-	validator.execSpawner = validatorclient.NewExecutionClient(valConfFetcher, stack)
-	return validator, nil
+	return nil
 }
 
 func (v *StatelessBlockValidator) GetModuleRootsToValidate() []common.Hash {

system_tests/block_validator_test.go

@@ -74,7 +74,6 @@ func testBlockValidatorSimple(t *testing.T, dasModeString string, workloadLoops
 	if useRedisStreams {
 		redisURL = redisutil.CreateTestRedis(ctx, t)
 		validatorConfig.BlockValidator.RedisValidationClientConfig = validatorclient.DefaultRedisValidationClientConfig
-		validatorConfig.BlockValidator.RedisValidationClientConfig.ModuleRoots = []string{currentRootModule(t).Hex()}
 		validatorConfig.BlockValidator.RedisValidationClientConfig.RedisURL = redisURL
 		validatorConfig.BlockValidator.ValidationServerConfigs = nil
 	}

validator/client/redisproducer.go

@@ -6,6 +6,8 @@ import (
 	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/go-redis/redis/v8"
 	"github.com/offchainlabs/nitro/pubsub"
 	"github.com/offchainlabs/nitro/util/containers"
 	"github.com/offchainlabs/nitro/util/redisutil"
@@ -21,8 +23,6 @@ type RedisValidationClientConfig struct {
 	Room           int32                 `koanf:"room"`
 	RedisURL       string                `koanf:"redis-url"`
 	ProducerConfig pubsub.ProducerConfig `koanf:"producer-config"`
-	// Supported wasm module roots, when the list is empty this is disabled.
-	ModuleRoots []string `koanf:"module-roots"`
 }
 
 func (c RedisValidationClientConfig) Enabled() bool {
@@ -47,7 +47,6 @@ func RedisValidationClientConfigAddOptions(prefix string, f *pflag.FlagSet) {
 	f.String(prefix+".name", DefaultRedisValidationClientConfig.Name, "validation client name")
 	f.Int32(prefix+".room", DefaultRedisValidationClientConfig.Room, "validation client room")
 	pubsub.ProducerAddConfigAddOptions(prefix+".producer-config", f)
-	f.StringSlice(prefix+".module-roots", nil, "Supported module root hashes")
 }
 
 // RedisValidationClient implements validation client through redis streams.
@@ -56,35 +55,43 @@ type RedisValidationClient struct {
 	name string
 	room int32
 	// producers stores moduleRoot to producer mapping.
-	producers map[common.Hash]*pubsub.Producer[*validator.ValidationInput, validator.GoGlobalState]
+	producers      map[common.Hash]*pubsub.Producer[*validator.ValidationInput, validator.GoGlobalState]
+	producerConfig pubsub.ProducerConfig
+	redisClient    redis.UniversalClient
 }
 
 func NewRedisValidationClient(cfg *RedisValidationClientConfig) (*RedisValidationClient, error) {
-	res := &RedisValidationClient{
-		name:      cfg.Name,
-		room:      cfg.Room,
-		producers: make(map[common.Hash]*pubsub.Producer[*validator.ValidationInput, validator.GoGlobalState]),
-	}
 	if cfg.RedisURL == "" {
 		return nil, fmt.Errorf("redis url cannot be empty")
 	}
 	redisClient, err := redisutil.RedisClientFromURL(cfg.RedisURL)
 	if err != nil {
 		return nil, err
 	}
-	if len(cfg.ModuleRoots) == 0 {
-		return nil, fmt.Errorf("moduleRoots must be specified to enable redis streams")
-	}
-	for _, hash := range cfg.ModuleRoots {
-		mr := common.HexToHash(hash)
+	return &RedisValidationClient{
+		name:           cfg.Name,
+		room:           cfg.Room,
+		producers:      make(map[common.Hash]*pubsub.Producer[*validator.ValidationInput, validator.GoGlobalState]),
+		producerConfig: cfg.ProducerConfig,
+		redisClient:    redisClient,
+	}, nil
+}
+
+func (c *RedisValidationClient) Initialize(moduleRoots []common.Hash) error {
+	for _, mr := range moduleRoots {
+		if _, exists := c.producers[mr]; exists {
+			log.Warn("Producer already existsw for module root", "hash", mr)
+			continue
+		}
 		p, err := pubsub.NewProducer[*validator.ValidationInput, validator.GoGlobalState](
-			redisClient, server_api.RedisStreamForRoot(mr), &cfg.ProducerConfig)
+			c.redisClient, server_api.RedisStreamForRoot(mr), &c.producerConfig)
 		if err != nil {
-			return nil, fmt.Errorf("creating producer for validation: %w", err)
+			return fmt.Errorf("creating producer for validation: %w", err)
 		}
-		res.producers[mr] = p
+		p.Start(c.GetContext())
+		c.producers[mr] = p
 	}
-	return res, nil
+	return nil
 }
 
 func (c *RedisValidationClient) Launch(entry *validator.ValidationInput, moduleRoot common.Hash) validator.ValidationRun {