[NIT-2740] Horizontal Scaling of Validation Node

staker/block_validator.go

@@ -91,21 +91,22 @@ type BlockValidator struct {
 }
 
 type BlockValidatorConfig struct {
-	Enable                      bool                          `koanf:"enable"`
-	Evil                        bool                          `koanf:"evil"`
-	EvilInterceptDepositGwei    uint64                        `koanf:"evil-intercept-deposit-gwei"`
-	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"`
+	Evil                            bool                             `koanf:"evil"`
+	EvilInterceptDepositGwei        uint64                           `koanf:"evil-intercept-deposit-gwei"`
+	RedisValidationClientConfig     redis.ValidationClientConfig     `koanf:"redis-validation-client-config"`
+	RedisBoldValidationClientConfig redis.BoldValidationClientConfig `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
 }
@@ -167,35 +168,37 @@ 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",
-	EvilInterceptDepositGwei:    1_000_000, // 1M gwei or 0.001 ETH
+	Enable:                          false,
+	ValidationServerConfigsList:     "default",
+	ValidationServer:                rpcclient.DefaultClientConfig,
+	RedisValidationClientConfig:     redis.DefaultValidationClientConfig,
+	RedisBoldValidationClientConfig: redis.DefaultBoldValidationClientConfig,
+	ValidationPoll:                  time.Second,
+	ForwardBlocks:                   1024,
+	PrerecordedBlocks:               uint64(2 * runtime.NumCPU()),
+	CurrentModuleRoot:               "current",
+	PendingUpgradeModuleRoot:        "latest",
+	FailureIsFatal:                  true,
+	Dangerous:                       DefaultBlockValidatorDangerousConfig,
+	MemoryFreeLimit:                 "default",
+	EvilInterceptDepositGwei:        1_000_000, // 1M gwei or 0.001 ETH
 }
 
 var TestBlockValidatorConfig = BlockValidatorConfig{
-	Enable:                      false,
-	EvilInterceptDepositGwei:    1_000_000, // 1M gwei or 0.001 ETH
-	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,
+	EvilInterceptDepositGwei:        1_000_000, // 1M gwei or 0.001 ETH
+	ValidationServer:                rpcclient.TestClientConfig,
+	ValidationServerConfigs:         []rpcclient.ClientConfig{rpcclient.TestClientConfig},
+	RedisValidationClientConfig:     redis.TestValidationClientConfig,
+	RedisBoldValidationClientConfig: redis.TestBoldValidationClientConfig,
+	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/stateless_block_validator.go

@@ -231,8 +231,11 @@
 	config func() *BlockValidatorConfig,
 	stack *node.Node,
 ) (*StatelessBlockValidator, error) {
-	var executionSpawners []validator.ExecutionSpawner
-	var redisValClient *redis.ValidationClient
+	var (
+		executionSpawners  []validator.ExecutionSpawner
+		redisValClient     *redis.ValidationClient
+		redisBoldValClient *redis.BoldValidationClient
+	)
 
 	if config().RedisValidationClientConfig.Enabled() {
 		var err error
@@ -241,11 +244,18 @@
 			return nil, fmt.Errorf("creating new redis validation client: %w", err)
 		}
 	}
+	if config().RedisBoldValidationClientConfig.Enabled() {
+		var err error
+		redisBoldValClient, err = redis.NewBoldValidationClient(&config().RedisBoldValidationClientConfig)
+		if err != nil {
+			return nil, fmt.Errorf("creating new redis bold validation client: %w", err)
+		}
+	}
 	configs := config().ValidationServerConfigs
 	for i := range configs {
 		i := i
 		confFetcher := func() *rpcclient.ClientConfig { return &config().ValidationServerConfigs[i] }
 		executionSpawners = append(executionSpawners, validatorclient.NewExecutionClient(confFetcher, stack))
 	}
 
 	if len(executionSpawners) == 0 {

system_tests/validation_mock_test.go

@@ -174,7 +174,7 @@
 	}
 	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,
@@ -205,7 +205,7 @@
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	_, validationDefault := createMockValidationNode(t, ctx, nil)
 	client := validatorclient.NewExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), validationDefault)
 	err := client.Start(ctx)
 	Require(t, err)
 
@@ -282,7 +282,7 @@
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	mockSpawner, spawnerStack := createMockValidationNode(t, ctx, nil)
 	client := validatorclient.NewExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), spawnerStack)
 	err := client.Start(ctx)
 	Require(t, err)
 

validator/client/redis/boldproducer.go

@@ -0,0 +1,106 @@
+package redis
+
+import (
+	"context"
+	"fmt"
+
+	"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"
+	"github.com/offchainlabs/nitro/util/stopwaiter"
+	"github.com/offchainlabs/nitro/validator/server_api"
+	"github.com/spf13/pflag"
+)
+
+type BoldValidationClientConfig struct {
+	RedisURL string `koanf:"redis-url"`
+}
+
+func (c BoldValidationClientConfig) Enabled() bool {
+	return c.RedisURL != ""
+}
+
+var DefaultBoldValidationClientConfig = BoldValidationClientConfig{
+	RedisURL: "",
+}
+
+var TestBoldValidationClientConfig = BoldValidationClientConfig{
+	RedisURL: "",
+}
+
+func BoldValidationClientConfigAddOptions(prefix string, f *pflag.FlagSet) {
+	pubsub.ProducerAddConfigAddOptions(prefix+".producer-config", f)
+}
+
+// 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]
+	producerConfig pubsub.ProducerConfig
+	redisClient    redis.UniversalClient
+	moduleRoots    []common.Hash
+}
+
+func NewBoldValidationClient(cfg *BoldValidationClientConfig) (*BoldValidationClient, error) {
+	if cfg.RedisURL == "" {
+		return nil, fmt.Errorf("redis url cannot be empty")
+	}
+	redisClient, err := redisutil.RedisClientFromURL(cfg.RedisURL)
+	if err != nil {
+		return nil, err
+	}
+	return &BoldValidationClient{
+		producers:   make(map[common.Hash]*pubsub.Producer[*server_api.GetLeavesWithStepSizeInput, []common.Hash]),
+		redisClient: redisClient,
+	}, nil
+}
+
+func (c *BoldValidationClient) 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[*server_api.GetLeavesWithStepSizeInput, []common.Hash](
+			c.redisClient, server_api.RedisStreamForRoot(mr), &c.producerConfig)
+		if err != nil {
+			log.Warn("failed init redis for %v: %w", mr, err)
+			continue
+		}
+		p.Start(c.GetContext())
+		c.producers[mr] = p
+		c.moduleRoots = append(c.moduleRoots, mr)
+	}
+	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"
 
@@ -115,11 +116,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.BoldValidationClientConfig, stack *node.Node) *ExecutionClient {
+	var boldClient *redis.BoldValidationClient
+	if 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,
 	}
 }
 
@@ -131,8 +142,10 @@ func (c *ExecutionClient) CreateBoldExecutionRun(wasmModuleRoot common.Hash, ste
 			return nil, err
 		}
 		run := &ExecutionClientRun{
-			client: c,
-			id:     res,
+			client:         c,
+			id:             res,
+			wasmModuleRoot: wasmModuleRoot,
+			input:          input,
 		}
 		run.Start(c.GetContext()) // note: not this temporary thread's context!
 		return run, nil
@@ -157,8 +170,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] {
@@ -213,6 +228,16 @@ func (r *ExecutionClientRun) GetStepAt(pos uint64) containers.PromiseInterface[*
 }
 
 func (r *ExecutionClientRun) GetLeavesWithStepSize(fromBatch, machineStartIndex, stepSize, numDesiredLeaves uint64) containers.PromiseInterface[[]common.Hash] {
+	if r.client.boldValClient != nil {
+		return r.client.boldValClient.GetLeavesWithStepSize(&server_api.GetLeavesWithStepSizeInput{
+			ModuleRoot:        r.wasmModuleRoot,
+			FromBatch:         fromBatch,
+			MachineStartIndex: machineStartIndex,
+			StepSize:          stepSize,
+			NumDesiredLeaves:  numDesiredLeaves,
+			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+"_getLeavesWithStepSize", r.id, fromBatch, machineStartIndex, stepSize, numDesiredLeaves)

validator/server_api/json.go

@@ -150,3 +150,12 @@ func ValidationInputFromJson(entry *InputJSON) (*validator.ValidationInput, erro
 	}
 	return valInput, nil
 }
+
+type GetLeavesWithStepSizeInput struct {
+	ModuleRoot        common.Hash
+	FromBatch         uint64
+	MachineStartIndex uint64
+	StepSize          uint64
+	NumDesiredLeaves  uint64
+	ValidationInput   *validator.ValidationInput
+}