[NIT-2740] Horizontal Scaling of Validation Node

staker/block_validator.go

@@ -100,19 +100,20 @@ type BlockValidator struct {
 }
 
 type BlockValidatorConfig struct {
-	Enable                      bool                          `koanf:"enable"`
-	RedisValidationClientConfig redis.ValidationClientConfig  `koanf:"redis-validation-client-config"`
-	ValidationServer            rpcclient.ClientConfig        `koanf:"validation-server" reload:"hot"`
-	ValidationServerConfigs     []rpcclient.ClientConfig      `koanf:"validation-server-configs"`
-	ValidationPoll              time.Duration                 `koanf:"validation-poll" reload:"hot"`
-	PrerecordedBlocks           uint64                        `koanf:"prerecorded-blocks" reload:"hot"`
-	ForwardBlocks               uint64                        `koanf:"forward-blocks" reload:"hot"`
-	CurrentModuleRoot           string                        `koanf:"current-module-root"`         // TODO(magic) requires reinitialization on hot reload
-	PendingUpgradeModuleRoot    string                        `koanf:"pending-upgrade-module-root"` // TODO(magic) requires StatelessBlockValidator recreation on hot reload
-	FailureIsFatal              bool                          `koanf:"failure-is-fatal" reload:"hot"`
-	Dangerous                   BlockValidatorDangerousConfig `koanf:"dangerous"`
-	MemoryFreeLimit             string                        `koanf:"memory-free-limit" reload:"hot"`
-	ValidationServerConfigsList string                        `koanf:"validation-server-configs-list"`
+	Enable                          bool                          `koanf:"enable"`
+	RedisValidationClientConfig     redis.ValidationClientConfig  `koanf:"redis-validation-client-config"`
+	RedisBoldValidationClientConfig redis.ValidationClientConfig  `koanf:"redis-bold-validation-client-config"`
+	ValidationServer                rpcclient.ClientConfig        `koanf:"validation-server" reload:"hot"`
+	ValidationServerConfigs         []rpcclient.ClientConfig      `koanf:"validation-server-configs"`
+	ValidationPoll                  time.Duration                 `koanf:"validation-poll" reload:"hot"`
+	PrerecordedBlocks               uint64                        `koanf:"prerecorded-blocks" reload:"hot"`
+	ForwardBlocks                   uint64                        `koanf:"forward-blocks" reload:"hot"`
+	CurrentModuleRoot               string                        `koanf:"current-module-root"`         // TODO(magic) requires reinitialization on hot reload
+	PendingUpgradeModuleRoot        string                        `koanf:"pending-upgrade-module-root"` // TODO(magic) requires StatelessBlockValidator recreation on hot reload
+	FailureIsFatal                  bool                          `koanf:"failure-is-fatal" reload:"hot"`
+	Dangerous                       BlockValidatorDangerousConfig `koanf:"dangerous"`
+	MemoryFreeLimit                 string                        `koanf:"memory-free-limit" reload:"hot"`
+	ValidationServerConfigsList     string                        `koanf:"validation-server-configs-list"`
 
 	memoryFreeLimit int
 }
@@ -169,6 +170,7 @@ 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)
 	redis.ValidationClientConfigAddOptions(prefix+".redis-validation-client-config", f)
+	redis.ValidationClientConfigAddOptions(prefix+".redis-bold-validation-client-config", f)
 	f.String(prefix+".validation-server-configs-list", DefaultBlockValidatorConfig.ValidationServerConfigsList, "array of execution 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)")
@@ -185,33 +187,35 @@ func BlockValidatorDangerousConfigAddOptions(prefix string, f *pflag.FlagSet) {
 }
 
 var DefaultBlockValidatorConfig = BlockValidatorConfig{
-	Enable:                      false,
-	ValidationServerConfigsList: "default",
-	ValidationServer:            rpcclient.DefaultClientConfig,
-	RedisValidationClientConfig: redis.DefaultValidationClientConfig,
-	ValidationPoll:              time.Second,
-	ForwardBlocks:               1024,
-	PrerecordedBlocks:           uint64(2 * runtime.NumCPU()),
-	CurrentModuleRoot:           "current",
-	PendingUpgradeModuleRoot:    "latest",
-	FailureIsFatal:              true,
-	Dangerous:                   DefaultBlockValidatorDangerousConfig,
-	MemoryFreeLimit:             "default",
+	Enable:                          false,
+	ValidationServerConfigsList:     "default",
+	ValidationServer:                rpcclient.DefaultClientConfig,
+	RedisValidationClientConfig:     redis.DefaultValidationClientConfig,
+	RedisBoldValidationClientConfig: redis.DefaultValidationClientConfig,
+	ValidationPoll:                  time.Second,
+	ForwardBlocks:                   1024,
+	PrerecordedBlocks:               uint64(2 * runtime.NumCPU()),
+	CurrentModuleRoot:               "current",
+	PendingUpgradeModuleRoot:        "latest",
+	FailureIsFatal:                  true,
+	Dangerous:                       DefaultBlockValidatorDangerousConfig,
+	MemoryFreeLimit:                 "default",
 }
 
 var TestBlockValidatorConfig = BlockValidatorConfig{
-	Enable:                      false,
-	ValidationServer:            rpcclient.TestClientConfig,
-	ValidationServerConfigs:     []rpcclient.ClientConfig{rpcclient.TestClientConfig},
-	RedisValidationClientConfig: redis.TestValidationClientConfig,
-	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:     redis.TestValidationClientConfig,
+	RedisBoldValidationClientConfig: redis.TestValidationClientConfig,
+	ValidationPoll:                  100 * time.Millisecond,
+	ForwardBlocks:                   128,
+	PrerecordedBlocks:               uint64(2 * runtime.NumCPU()),
+	CurrentModuleRoot:               "latest",
+	PendingUpgradeModuleRoot:        "latest",
+	FailureIsFatal:                  true,
+	Dangerous:                       DefaultBlockValidatorDangerousConfig,
+	MemoryFreeLimit:                 "default",
 }
 
 var DefaultBlockValidatorDangerousConfig = BlockValidatorDangerousConfig{

staker/bold_state_provider.go

@@ -375,7 +375,6 @@ func (s *BOLDStateProvider) CollectMachineHashes(
 	if err != nil {
 		return nil, err
 	}
-	// TODO: Enable Redis streams.
 	execRun, err := s.statelessValidator.execSpawners[0].CreateExecutionRun(cfg.WasmModuleRoot, input).Await(ctx)
 	if err != nil {
 		return nil, err

staker/stateless_block_validator.go

@@ -228,7 +228,10 @@ func NewStatelessBlockValidator(
 	for i := range configs {
 		i := i
 		confFetcher := func() *rpcclient.ClientConfig { return &config().ValidationServerConfigs[i] }
-		executionSpawners = append(executionSpawners, validatorclient.NewExecutionClient(confFetcher, stack))
+		executionSpawners = append(executionSpawners, validatorclient.NewExecutionClient(confFetcher, nil, stack))
+		if i == 0 {
+			executionSpawners = append(executionSpawners, validatorclient.NewExecutionClient(confFetcher, &config().RedisBoldValidationClientConfig, stack))
+		}
 	}
 
 	if len(executionSpawners) == 0 {

system_tests/validation_mock_test.go

@@ -184,7 +184,7 @@ func createMockValidationNode(t *testing.T, ctx context.Context, config *server_
 	}
 	configFetcher := func() *server_arb.ArbitratorSpawnerConfig { return config }
 	spawner := &mockSpawner{}
-	serverAPI := valnode.NewExecutionServerAPI(spawner, spawner, configFetcher)
+	serverAPI := valnode.NewExecutionServerAPI(spawner, spawner, nil, configFetcher)
 
 	valAPIs := []rpc.API{{
 		Namespace:     server_api.Namespace,
@@ -215,7 +215,7 @@ func TestValidationServerAPI(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	_, validationDefault := createMockValidationNode(t, ctx, nil)
-	client := validatorclient.NewExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), validationDefault)
+	client := validatorclient.NewExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), nil, validationDefault)
 	err := client.Start(ctx)
 	Require(t, err)
 
@@ -292,7 +292,7 @@ func TestValidationClientRoom(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	mockSpawner, spawnerStack := createMockValidationNode(t, ctx, nil)
-	client := validatorclient.NewExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), spawnerStack)
+	client := validatorclient.NewExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), nil, spawnerStack)
 	err := client.Start(ctx)
 	Require(t, err)
 
@@ -379,10 +379,10 @@ func TestExecutionKeepAlive(t *testing.T) {
 	_, validationShortTO := createMockValidationNode(t, ctx, &shortTimeoutConfig)
 	configFetcher := StaticFetcherFrom(t, &rpcclient.TestClientConfig)
 
-	clientDefault := validatorclient.NewExecutionClient(configFetcher, validationDefault)
+	clientDefault := validatorclient.NewExecutionClient(configFetcher, nil, validationDefault)
 	err := clientDefault.Start(ctx)
 	Require(t, err)
-	clientShortTO := validatorclient.NewExecutionClient(configFetcher, validationShortTO)
+	clientShortTO := validatorclient.NewExecutionClient(configFetcher, nil, validationShortTO)
 	err = clientShortTO.Start(ctx)
 	Require(t, err)
 

validator/client/redis/boldproducer.go

@@ -0,0 +1,86 @@
+package redis
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/offchainlabs/nitro/pubsub"
+	"github.com/offchainlabs/nitro/util/containers"
+	"github.com/offchainlabs/nitro/util/redisutil"
+	"github.com/offchainlabs/nitro/util/stopwaiter"
+	"github.com/offchainlabs/nitro/validator/server_api"
+)
+
+// BoldValidationClient implements bold validation client through redis streams.
+type BoldValidationClient struct {
+	stopwaiter.StopWaiter
+	// producers stores moduleRoot to producer mapping.
+	producers map[common.Hash]*pubsub.Producer[*server_api.GetLeavesWithStepSizeInput, []common.Hash]
+	config    *ValidationClientConfig
+}
+
+func NewBoldValidationClient(cfg *ValidationClientConfig) (*BoldValidationClient, error) {
+	return &BoldValidationClient{
+		producers: make(map[common.Hash]*pubsub.Producer[*server_api.GetLeavesWithStepSizeInput, []common.Hash]),
+		config:    cfg,
+	}, nil
+}
+
+func (c *BoldValidationClient) Initialize(ctx context.Context, moduleRoots []common.Hash) error {
+	if c.config.RedisURL == "" {
+		return fmt.Errorf("redis url cannot be empty")
+	}
+	redisClient, err := redisutil.RedisClientFromURL(c.config.RedisURL)
+	if err != nil {
+		return err
+	}
+	for _, mr := range moduleRoots {
+		if c.config.CreateStreams {
+			if err := pubsub.CreateStream(ctx, server_api.RedisBoldStreamForRoot(c.config.StreamPrefix, mr), redisClient); err != nil {
+				return fmt.Errorf("creating redis stream: %w", err)
+			}
+		}
+		if _, exists := c.producers[mr]; exists {
+			log.Warn("Producer already existsw for module root", "hash", mr)
+			continue
+		}
+		p, err := pubsub.NewProducer[*server_api.GetLeavesWithStepSizeInput, []common.Hash](
+			redisClient, server_api.RedisBoldStreamForRoot(c.config.StreamPrefix, mr), &c.config.ProducerConfig)
+		if err != nil {
+			log.Warn("failed init redis for %v: %w", mr, err)
+			continue
+		}
+		p.Start(c.GetContext())
+		c.producers[mr] = p
+	}
+	return nil
+}
+
+func (c *BoldValidationClient) GetLeavesWithStepSize(req *server_api.GetLeavesWithStepSizeInput) containers.PromiseInterface[[]common.Hash] {
+	producer, found := c.producers[req.ModuleRoot]
+	if !found {
+		return containers.NewReadyPromise([]common.Hash{}, fmt.Errorf("no validation is configured for wasm root %v", req.ModuleRoot))
+	}
+	promise, err := producer.Produce(c.GetContext(), req)
+	if err != nil {
+		return containers.NewReadyPromise([]common.Hash{}, fmt.Errorf("error producing input: %w", err))
+	}
+	return promise
+}
+
+func (c *BoldValidationClient) Start(ctx_in context.Context) error {
+	for _, p := range c.producers {
+		p.Start(ctx_in)
+	}
+	c.StopWaiter.Start(ctx_in, c)
+	return nil
+}
+
+func (c *BoldValidationClient) Stop() {
+	for _, p := range c.producers {
+		p.StopAndWait()
+	}
+	c.StopWaiter.StopAndWait()
+}

validator/client/validation_client.go

@@ -17,6 +17,7 @@ import (
 	"github.com/offchainlabs/nitro/util/rpcclient"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
 
+	"github.com/offchainlabs/nitro/validator/client/redis"
 	"github.com/offchainlabs/nitro/validator/server_api"
 	"github.com/offchainlabs/nitro/validator/server_common"
 
@@ -146,11 +147,21 @@ func (c *ValidationClient) Room() int {
 
 type ExecutionClient struct {
 	ValidationClient
+	boldValClient *redis.BoldValidationClient
 }
 
-func NewExecutionClient(config rpcclient.ClientConfigFetcher, stack *node.Node) *ExecutionClient {
+func NewExecutionClient(config rpcclient.ClientConfigFetcher, redisBoldValidationClientConfig *redis.ValidationClientConfig, stack *node.Node) *ExecutionClient {
+	var boldClient *redis.BoldValidationClient
+	if redisBoldValidationClientConfig != nil && redisBoldValidationClientConfig.Enabled() {
+		var err error
+		boldClient, err = redis.NewBoldValidationClient(redisBoldValidationClientConfig)
+		if err != nil {
+			log.Error("Creating new redis bold validation client", "error", err)
+		}
+	}
 	return &ExecutionClient{
 		ValidationClient: *NewValidationClient(config, stack),
+		boldValClient:    boldClient,
 	}
 }
 
@@ -172,8 +183,10 @@ func (c *ExecutionClient) CreateExecutionRun(wasmModuleRoot common.Hash, input *
 
 type ExecutionClientRun struct {
 	stopwaiter.StopWaiter
-	client *ExecutionClient
-	id     uint64
+	client         *ExecutionClient
+	id             uint64
+	wasmModuleRoot common.Hash
+	input          *validator.ValidationInput
 }
 
 func (c *ExecutionClient) LatestWasmModuleRoot() containers.PromiseInterface[common.Hash] {
@@ -233,6 +246,15 @@ func (r *ExecutionClientRun) GetStepAt(pos uint64) containers.PromiseInterface[*
 }
 
 func (r *ExecutionClientRun) GetMachineHashesWithStepSize(machineStartIndex, stepSize, maxIterations uint64) containers.PromiseInterface[[]common.Hash] {
+	if r.client.boldValClient != nil {
+		return r.client.boldValClient.GetLeavesWithStepSize(&server_api.GetLeavesWithStepSizeInput{
+			ModuleRoot:        r.wasmModuleRoot,
+			MachineStartIndex: machineStartIndex,
+			StepSize:          stepSize,
+			NumDesiredLeaves:  maxIterations,
+			ValidationInput:   r.input,
+		})
+	}
 	return stopwaiter.LaunchPromiseThread[[]common.Hash](r, func(ctx context.Context) ([]common.Hash, error) {
 		var resJson []common.Hash
 		err := r.client.client.CallContext(ctx, &resJson, server_api.Namespace+"_getMachineHashesWithStepSize", r.id, machineStartIndex, stepSize, maxIterations)

validator/server_api/json.go

@@ -51,6 +51,10 @@ func RedisStreamForRoot(prefix string, moduleRoot common.Hash) string {
 	return fmt.Sprintf("%sstream:%s", prefix, moduleRoot.Hex())
 }
 
+func RedisBoldStreamForRoot(prefix string, moduleRoot common.Hash) string {
+	return fmt.Sprintf("%sstream-bold:%s", prefix, moduleRoot.Hex())
+}
+
 type Request struct {
 	Input      *InputJSON
 	ModuleRoot common.Hash
@@ -176,3 +180,11 @@ func ValidationInputFromJson(entry *InputJSON) (*validator.ValidationInput, erro
 	}
 	return valInput, nil
 }
+
+type GetLeavesWithStepSizeInput struct {
+	ModuleRoot        common.Hash
+	MachineStartIndex uint64
+	StepSize          uint64
+	NumDesiredLeaves  uint64
+	ValidationInput   *validator.ValidationInput
+}