[NIT-2740] Horizontal Scaling of Validation Node

staker/bold/bold_state_provider.go

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

staker/stateless_block_validator.go

@@ -250,6 +250,11 @@ func NewStatelessBlockValidator(
 		i := i
 		confFetcher := func() *rpcclient.ClientConfig { return &config().ValidationServerConfigs[i] }
 		executionSpawners = append(executionSpawners, validatorclient.NewExecutionClient(confFetcher, stack))
+		if i == 0 {
+			if config().RedisValidationClientConfig.Enabled() {
+				executionSpawners = append(executionSpawners, validatorclient.NewBoldExecutionClient(confFetcher, &config().RedisValidationClientConfig, stack))
+			}
+		}
 	}
 
 	if len(executionSpawners) == 0 {

system_tests/validation_mock_test.go

@@ -20,9 +20,11 @@ import (
 	"github.com/offchainlabs/nitro/execution"
 	"github.com/offchainlabs/nitro/staker"
 	"github.com/offchainlabs/nitro/util/containers"
+	"github.com/offchainlabs/nitro/util/redisutil"
 	"github.com/offchainlabs/nitro/util/rpcclient"
 	"github.com/offchainlabs/nitro/validator"
 	validatorclient "github.com/offchainlabs/nitro/validator/client"
+	clientredis "github.com/offchainlabs/nitro/validator/client/redis"
 	"github.com/offchainlabs/nitro/validator/server_api"
 	"github.com/offchainlabs/nitro/validator/server_arb"
 	"github.com/offchainlabs/nitro/validator/valnode"
@@ -206,11 +208,26 @@ func createMockValidationNode(t *testing.T, ctx context.Context, config *server_
 
 // mostly tests translation to/from json and running over network
 func TestValidationServerAPI(t *testing.T) {
+	testValidationServerAPI(t, false)
+}
+
+// mostly tests translation to/from json and running over network with bold validation redis consumer/producer
+func TestValidationServerAPIWithBoldValidationConsumerProducer(t *testing.T) {
+	testValidationServerAPI(t, true)
+}
+func testValidationServerAPI(t *testing.T, withBoldValidationConsumerProducer bool) {
 	t.Parallel()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+	var redisBoldValidationClientConfig *clientredis.ValidationClientConfig
+	if withBoldValidationConsumerProducer {
+		redisBoldValidationClientConfig = &clientredis.TestValidationClientConfig
+		redisBoldValidationClientConfig.RedisURL = redisutil.CreateTestRedis(ctx, t)
+		redisBoldValidationClientConfig.CreateStreams = true
+	}
+
 	_, validationDefault := createMockValidationNode(t, ctx, nil)
-	client := validatorclient.NewExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), validationDefault)
+	client := validatorclient.NewBoldExecutionClient(StaticFetcherFrom(t, &rpcclient.TestClientConfig), redisBoldValidationClientConfig, validationDefault)
 	err := client.Start(ctx)
 	Require(t, err)
 
@@ -280,6 +297,13 @@ func TestValidationServerAPI(t *testing.T) {
 	if !bytes.Equal(proof, mockProof) {
 		t.Error("mock proof not expected")
 	}
+
+	hashes := execRun.GetMachineHashesWithStepSize(0, 1, 5)
+	hashesRes, err := hashes.Await(ctx)
+	Require(t, err)
+	if len(hashesRes) != 5 {
+		t.Error("unexpected number of hashes")
+	}
 }
 
 func TestValidationClientRoom(t *testing.T) {

validator/client/validation_client.go

@@ -18,10 +18,13 @@ import (
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/rpc"
 
+	"github.com/offchainlabs/nitro/pubsub"
 	"github.com/offchainlabs/nitro/util/containers"
+	"github.com/offchainlabs/nitro/util/redisutil"
 	"github.com/offchainlabs/nitro/util/rpcclient"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
 	"github.com/offchainlabs/nitro/validator"
+	"github.com/offchainlabs/nitro/validator/client/redis"
 	"github.com/offchainlabs/nitro/validator/server_api"
 	"github.com/offchainlabs/nitro/validator/server_common"
 )
@@ -144,14 +147,20 @@ func (c *ValidationClient) Room() int {
 }
 
 type ExecutionClient struct {
-	ValidationClient
+	*ValidationClient
 }
 
 func NewExecutionClient(config rpcclient.ClientConfigFetcher, stack *node.Node) *ExecutionClient {
 	return &ExecutionClient{
-		ValidationClient: *NewValidationClient(config, stack),
+		ValidationClient: NewValidationClient(config, stack),
 	}
 }
+func (c *ExecutionClient) Start(ctx context.Context) error {
+	if err := c.ValidationClient.Start(ctx); err != nil {
+		return err
+	}
+	return nil
+}
 
 func (c *ExecutionClient) CreateExecutionRun(
 	wasmModuleRoot common.Hash,
@@ -165,8 +174,10 @@ func (c *ExecutionClient) CreateExecutionRun(
 			return nil, err
 		}
 		run := &ExecutionClientRun{
-			client: c,
-			id:     res,
+			wasmModuleRoot: wasmModuleRoot,
+			client:         c,
+			id:             res,
+			input:          input,
 		}
 		run.Start(c.GetContext()) // note: not this temporary thread's context!
 		return run, nil
@@ -175,8 +186,11 @@ func (c *ExecutionClient) CreateExecutionRun(
 
 type ExecutionClientRun struct {
 	stopwaiter.StopWaiter
-	client *ExecutionClient
-	id     uint64
+	client         *ExecutionClient
+	boldClient     *BoldExecutionClient
+	id             uint64
+	wasmModuleRoot common.Hash
+	input          *validator.ValidationInput
 }
 
 func (c *ExecutionClient) LatestWasmModuleRoot() containers.PromiseInterface[common.Hash] {
@@ -223,6 +237,15 @@ func (r *ExecutionClientRun) GetStepAt(pos uint64) containers.PromiseInterface[*
 }
 
 func (r *ExecutionClientRun) GetMachineHashesWithStepSize(machineStartIndex, stepSize, maxIterations uint64) containers.PromiseInterface[[]common.Hash] {
+	if r.boldClient != nil {
+		return r.boldClient.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)
@@ -267,3 +290,131 @@ func (r *ExecutionClientRun) Close() {
 		}
 	})
 }
+
+// BoldValidationClient implements bold validation client through redis streams.
+type BoldValidationClient struct {
+	stopwaiter.StopWaiter
+	*ValidationClient
+	// producers stores moduleRoot to producer mapping.
+	producers map[common.Hash]*pubsub.Producer[*server_api.GetLeavesWithStepSizeInput, []common.Hash]
+	config    *redis.ValidationClientConfig
+}
+
+func NewBoldValidationClient(config rpcclient.ClientConfigFetcher, redisBoldValidationClientConfig *redis.ValidationClientConfig, stack *node.Node) (*BoldValidationClient, error) {
+	return &BoldValidationClient{
+		ValidationClient: NewValidationClient(config, stack),
+		producers:        make(map[common.Hash]*pubsub.Producer[*server_api.GetLeavesWithStepSizeInput, []common.Hash]),
+		config:           redisBoldValidationClientConfig,
+	}, 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 exists 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
+		}
+		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 {
+	if err := c.Initialize(ctx_in, c.wasmModuleRoots); err != nil {
+		return err
+	}
+	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()
+}
+
+type BoldExecutionClient struct {
+	*BoldValidationClient
+}
+
+func NewBoldExecutionClient(config rpcclient.ClientConfigFetcher, redisBoldValidationClientConfig *redis.ValidationClientConfig, stack *node.Node) *BoldExecutionClient {
+	var validationClient *BoldValidationClient
+	if redisBoldValidationClientConfig != nil && redisBoldValidationClientConfig.Enabled() {
+		var err error
+		validationClient, err = NewBoldValidationClient(config, redisBoldValidationClientConfig, stack)
+		if err != nil {
+			log.Error("Creating new redis bold validation client", "error", err)
+		}
+	}
+	return &BoldExecutionClient{
+		BoldValidationClient: validationClient,
+	}
+}
+
+func (c *BoldExecutionClient) CreateExecutionRun(
+	wasmModuleRoot common.Hash,
+	input *validator.ValidationInput,
+	useBoldMachine bool,
+) containers.PromiseInterface[validator.ExecutionRun] {
+	return stopwaiter.LaunchPromiseThread(c, func(ctx context.Context) (validator.ExecutionRun, error) {
+		var res uint64
+		err := c.client.CallContext(ctx, &res, server_api.Namespace+"_createExecutionRun", wasmModuleRoot, server_api.ValidationInputToJson(input), useBoldMachine)
+		if err != nil {
+			return nil, err
+		}
+		run := &ExecutionClientRun{
+			wasmModuleRoot: wasmModuleRoot,
+			boldClient:     c,
+			client:         &ExecutionClient{ValidationClient: c.BoldValidationClient.ValidationClient},
+			id:             res,
+			input:          input,
+		}
+		run.Start(c.GetContext()) // note: not this temporary thread's context!
+		return run, nil
+	})
+}
+
+func (c *BoldExecutionClient) LatestWasmModuleRoot() containers.PromiseInterface[common.Hash] {
+	return stopwaiter.LaunchPromiseThread[common.Hash](c, func(ctx context.Context) (common.Hash, error) {
+		var res common.Hash
+		err := c.client.CallContext(ctx, &res, server_api.Namespace+"_latestWasmModuleRoot")
+		if err != nil {
+			return common.Hash{}, err
+		}
+		return res, nil
+	})
+}

validator/server_api/json.go

@@ -50,6 +50,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
@@ -169,3 +173,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
+}