Integrate Redis streams in to Nitro's validation

arbnode/node.go

@@ -540,7 +540,7 @@ func createNodeImpl(
 	txStreamer.SetInboxReaders(inboxReader, delayedBridge)
 
 	var statelessBlockValidator *staker.StatelessBlockValidator
-	if config.BlockValidator.ValidationServerConfigs[0].URL != "" {
+	if config.BlockValidator.RedisValidationClientConfig.Enabled() || config.BlockValidator.ValidationServerConfigs[0].URL != "" {
 		statelessBlockValidator, err = staker.NewStatelessBlockValidator(
 			inboxReader,
 			inboxTracker,

pubsub/consumer.go

@@ -10,7 +10,6 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/go-redis/redis/v8"
 	"github.com/google/uuid"
-	"github.com/offchainlabs/nitro/util/redisutil"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
 	"github.com/spf13/pflag"
 )
@@ -21,68 +20,49 @@ type ConsumerConfig struct {
 	// Duration after which consumer is considered to be dead if heartbeat
 	// is not updated.
 	KeepAliveTimeout time.Duration `koanf:"keepalive-timeout"`
-	// Redis url for Redis streams and locks.
-	RedisURL string `koanf:"redis-url"`
-	// Redis stream name.
-	RedisStream string `koanf:"redis-stream"`
-	// Redis consumer group name.
-	RedisGroup string `koanf:"redis-group"`
 }
 
-var DefaultConsumerConfig = &ConsumerConfig{
+var DefaultConsumerConfig = ConsumerConfig{
 	ResponseEntryTimeout: time.Hour,
 	KeepAliveTimeout:     5 * time.Minute,
-	RedisStream:          "",
-	RedisGroup:           "",
 }
 
-var DefaultTestConsumerConfig = &ConsumerConfig{
-	RedisStream:          "",
-	RedisGroup:           "",
+var TestConsumerConfig = ConsumerConfig{
 	ResponseEntryTimeout: time.Minute,
 	KeepAliveTimeout:     30 * time.Millisecond,
 }
 
 func ConsumerConfigAddOptions(prefix string, f *pflag.FlagSet) {
 	f.Duration(prefix+".response-entry-timeout", DefaultConsumerConfig.ResponseEntryTimeout, "timeout for response entry")
 	f.Duration(prefix+".keepalive-timeout", DefaultConsumerConfig.KeepAliveTimeout, "timeout after which consumer is considered inactive if heartbeat wasn't performed")
-	f.String(prefix+".redis-url", DefaultConsumerConfig.RedisURL, "redis url for redis stream")
-	f.String(prefix+".redis-stream", DefaultConsumerConfig.RedisStream, "redis stream name to read from")
-	f.String(prefix+".redis-group", DefaultConsumerConfig.RedisGroup, "redis stream consumer group name")
 }
 
 // Consumer implements a consumer for redis stream provides heartbeat to
 // indicate it is alive.
 type Consumer[Request any, Response any] struct {
 	stopwaiter.StopWaiter
-	id     string
-	client redis.UniversalClient
-	cfg    *ConsumerConfig
+	id          string
+	client      redis.UniversalClient
+	redisStream string
+	redisGroup  string
+	cfg         *ConsumerConfig
 }
 
 type Message[Request any] struct {
 	ID    string
 	Value Request
 }
 
-func NewConsumer[Request any, Response any](ctx context.Context, cfg *ConsumerConfig) (*Consumer[Request, Response], error) {
-	if cfg.RedisURL == "" {
-		return nil, fmt.Errorf("redis url cannot be empty")
-	}
-	if cfg.RedisStream == "" {
+func NewConsumer[Request any, Response any](client redis.UniversalClient, streamName string, cfg *ConsumerConfig) (*Consumer[Request, Response], error) {
+	if streamName == "" {
 		return nil, fmt.Errorf("redis stream name cannot be empty")
 	}
-	if cfg.RedisGroup == "" {
-		return nil, fmt.Errorf("redis group name cannot be emtpy")
-	}
-	c, err := redisutil.RedisClientFromURL(cfg.RedisURL)
-	if err != nil {
-		return nil, err
-	}
 	consumer := &Consumer[Request, Response]{
-		id:     uuid.NewString(),
-		client: c,
-		cfg:    cfg,
+		id:          uuid.NewString(),
+		client:      client,
+		redisStream: streamName,
+		redisGroup:  streamName, // There is 1-1 mapping of redis stream and consumer group.
+		cfg:         cfg,
 	}
 	return consumer, nil
 }
@@ -125,11 +105,11 @@ func (c *Consumer[Request, Response]) heartBeat(ctx context.Context) {
 // unresponsive consumer, if not then reads from the stream.
 func (c *Consumer[Request, Response]) Consume(ctx context.Context) (*Message[Request], error) {
 	res, err := c.client.XReadGroup(ctx, &redis.XReadGroupArgs{
-		Group:    c.cfg.RedisGroup,
+		Group:    c.redisGroup,
 		Consumer: c.id,
 		// Receive only messages that were never delivered to any other consumer,
 		// that is, only new messages.
-		Streams: []string{c.cfg.RedisStream, ">"},
+		Streams: []string{c.redisStream, ">"},
 		Count:   1,
 		Block:   time.Millisecond, // 0 seems to block the read instead of immediately returning
 	}).Result()
@@ -142,7 +122,6 @@ func (c *Consumer[Request, Response]) Consume(ctx context.Context) (*Message[Req
 	if len(res) != 1 || len(res[0].Messages) != 1 {
 		return nil, fmt.Errorf("redis returned entries: %+v, for querying single message", res)
 	}
-	log.Debug(fmt.Sprintf("Consumer: %s consuming message: %s", c.id, res[0].Messages[0].ID))
 	var (
 		value    = res[0].Messages[0].Values[messageKey]
 		data, ok = (value).(string)
@@ -154,7 +133,7 @@ func (c *Consumer[Request, Response]) Consume(ctx context.Context) (*Message[Req
 	if err := json.Unmarshal([]byte(data), &req); err != nil {
 		return nil, fmt.Errorf("unmarshaling value: %v, error: %w", value, err)
 	}
-
+	log.Debug("Redis stream consuming", "consumer_id", c.id, "message_id", res[0].Messages[0].ID)
 	return &Message[Request]{
 		ID:    res[0].Messages[0].ID,
 		Value: req,
@@ -170,7 +149,7 @@ func (c *Consumer[Request, Response]) SetResult(ctx context.Context, messageID s
 	if err != nil || !acquired {
 		return fmt.Errorf("setting result for  message: %v, error: %w", messageID, err)
 	}
-	if _, err := c.client.XAck(ctx, c.cfg.RedisStream, c.cfg.RedisGroup, messageID).Result(); err != nil {
+	if _, err := c.client.XAck(ctx, c.redisStream, c.redisGroup, messageID).Result(); err != nil {
 		return fmt.Errorf("acking message: %v, error: %w", messageID, err)
 	}
 	return nil

pubsub/producer.go

@@ -20,7 +20,6 @@ import (
 	"github.com/go-redis/redis/v8"
 	"github.com/google/uuid"
 	"github.com/offchainlabs/nitro/util/containers"
-	"github.com/offchainlabs/nitro/util/redisutil"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
 	"github.com/spf13/pflag"
 )
@@ -32,9 +31,11 @@ const (
 
 type Producer[Request any, Response any] struct {
 	stopwaiter.StopWaiter
-	id     string
-	client redis.UniversalClient
-	cfg    *ProducerConfig
+	id          string
+	client      redis.UniversalClient
+	redisStream string
+	redisGroup  string
+	cfg         *ProducerConfig
 
 	promisesLock sync.RWMutex
 	promises     map[string]*containers.Promise[Response]
@@ -49,10 +50,7 @@ type ProducerConfig struct {
 	// When enabled, messages that are sent to consumers that later die before
 	// processing them, will be re-inserted into the stream to be proceesed by
 	// another consumer
-	EnableReproduce bool   `koanf:"enable-reproduce"`
-	RedisURL        string `koanf:"redis-url"`
-	// Redis stream name.
-	RedisStream string `koanf:"redis-stream"`
+	EnableReproduce bool `koanf:"enable-reproduce"`
 	// Interval duration in which producer checks for pending messages delivered
 	// to the consumers that are currently inactive.
 	CheckPendingInterval time.Duration `koanf:"check-pending-interval"`
@@ -61,56 +59,43 @@ type ProducerConfig struct {
 	KeepAliveTimeout time.Duration `koanf:"keepalive-timeout"`
 	// Interval duration for checking the result set by consumers.
 	CheckResultInterval time.Duration `koanf:"check-result-interval"`
-	// Redis consumer group name.
-	RedisGroup string `koanf:"redis-group"`
 }
 
-var DefaultProducerConfig = &ProducerConfig{
+var DefaultProducerConfig = ProducerConfig{
 	EnableReproduce:      true,
-	RedisStream:          "",
-	RedisGroup:           "",
 	CheckPendingInterval: time.Second,
 	KeepAliveTimeout:     5 * time.Minute,
 	CheckResultInterval:  5 * time.Second,
 }
 
-var DefaultTestProducerConfig = &ProducerConfig{
+var TestProducerConfig = ProducerConfig{
 	EnableReproduce:      true,
-	RedisStream:          "",
-	RedisGroup:           "",
 	CheckPendingInterval: 10 * time.Millisecond,
 	KeepAliveTimeout:     100 * time.Millisecond,
 	CheckResultInterval:  5 * time.Millisecond,
 }
 
 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.String(prefix+".redis-url", DefaultProducerConfig.RedisURL, "redis url for redis 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")
-	f.String(prefix+".redis-stream", DefaultProducerConfig.RedisStream, "redis stream name to read from")
-	f.String(prefix+".redis-group", DefaultProducerConfig.RedisGroup, "redis stream consumer group name")
 }
 
-func NewProducer[Request any, Response any](cfg *ProducerConfig) (*Producer[Request, Response], error) {
-	if cfg.RedisURL == "" {
-		return nil, fmt.Errorf("redis url cannot be empty")
+func NewProducer[Request any, Response any](client redis.UniversalClient, streamName string, cfg *ProducerConfig) (*Producer[Request, Response], error) {
+	if client == nil {
+		return nil, fmt.Errorf("redis client cannot be nil")
 	}
-	if cfg.RedisStream == "" {
-		return nil, fmt.Errorf("redis stream cannot be emtpy")
-	}
-	if cfg.RedisGroup == "" {
-		return nil, fmt.Errorf("redis group cannot be empty")
-	}
-	c, err := redisutil.RedisClientFromURL(cfg.RedisURL)
-	if err != nil {
-		return nil, err
+	if streamName == "" {
+		return nil, fmt.Errorf("stream name cannot be empty")
 	}
 	return &Producer[Request, Response]{
-		id:       uuid.NewString(),
-		client:   c,
-		cfg:      cfg,
-		promises: make(map[string]*containers.Promise[Response]),
+		id:          uuid.NewString(),
+		client:      client,
+		redisStream: streamName,
+		redisGroup:  streamName, // There is 1-1 mapping of redis stream and consumer group.
+		cfg:         cfg,
+		promises:    make(map[string]*containers.Promise[Response]),
 	}, nil
 }
 
@@ -142,7 +127,7 @@ func (p *Producer[Request, Response]) checkAndReproduce(ctx context.Context) tim
 	}
 	acked := make(map[string]Request)
 	for _, msg := range msgs {
-		if _, err := p.client.XAck(ctx, p.cfg.RedisStream, p.cfg.RedisGroup, msg.ID).Result(); err != nil {
+		if _, err := p.client.XAck(ctx, p.redisStream, p.redisGroup, msg.ID).Result(); err != nil {
 			log.Error("ACKing message", "error", err)
 			continue
 		}
@@ -200,7 +185,7 @@ func (p *Producer[Request, Response]) reproduce(ctx context.Context, value Reque
 		return nil, fmt.Errorf("marshaling value: %w", err)
 	}
 	id, err := p.client.XAdd(ctx, &redis.XAddArgs{
-		Stream: p.cfg.RedisStream,
+		Stream: p.redisStream,
 		Values: map[string]any{messageKey: val},
 	}).Result()
 	if err != nil {
@@ -224,6 +209,7 @@ func (p *Producer[Request, Response]) reproduce(ctx context.Context, value Reque
 }
 
 func (p *Producer[Request, Response]) Produce(ctx context.Context, value Request) (*containers.Promise[Response], error) {
+	log.Debug("Redis stream producing", "value", value)
 	p.once.Do(func() {
 		p.StopWaiter.CallIteratively(p.checkAndReproduce)
 		p.StopWaiter.CallIteratively(p.checkResponses)
@@ -248,8 +234,8 @@ func (p *Producer[Request, Response]) havePromiseFor(messageID string) bool {
 
 func (p *Producer[Request, Response]) checkPending(ctx context.Context) ([]*Message[Request], error) {
 	pendingMessages, err := p.client.XPendingExt(ctx, &redis.XPendingExtArgs{
-		Stream: p.cfg.RedisStream,
-		Group:  p.cfg.RedisGroup,
+		Stream: p.redisStream,
+		Group:  p.redisGroup,
 		Start:  "-",
 		End:    "+",
 		Count:  100,
@@ -285,8 +271,8 @@ func (p *Producer[Request, Response]) checkPending(ctx context.Context) ([]*Mess
 	}
 	log.Info("Attempting to claim", "messages", ids)
 	claimedMsgs, err := p.client.XClaim(ctx, &redis.XClaimArgs{
-		Stream:   p.cfg.RedisStream,
-		Group:    p.cfg.RedisGroup,
+		Stream:   p.redisStream,
+		Group:    p.redisGroup,
 		Consumer: p.id,
 		MinIdle:  p.cfg.KeepAliveTimeout,
 		Messages: ids,

pubsub/pubsub_test.go

@@ -28,16 +28,17 @@ type testResponse struct {
 	Response string
 }
 
-func createGroup(ctx context.Context, t *testing.T, streamName, groupName string, client redis.UniversalClient) {
+func createRedisGroup(ctx context.Context, t *testing.T, streamName string, client redis.UniversalClient) {
 	t.Helper()
-	if _, err := client.XGroupCreateMkStream(ctx, streamName, groupName, "$").Result(); err != nil {
+	// Stream name and group name are the same.
+	if _, err := client.XGroupCreateMkStream(ctx, streamName, streamName, "$").Result(); err != nil {
 		t.Fatalf("Error creating stream group: %v", err)
 	}
 }
 
-func destroyGroup(ctx context.Context, t *testing.T, streamName, groupName string, client redis.UniversalClient) {
+func destroyRedisGroup(ctx context.Context, t *testing.T, streamName string, client redis.UniversalClient) {
 	t.Helper()
-	if _, err := client.XGroupDestroy(ctx, streamName, groupName).Result(); err != nil {
+	if _, err := client.XGroupDestroy(ctx, streamName, streamName).Result(); err != nil {
 		log.Debug("Error destroying a stream group", "error", err)
 	}
 }
@@ -54,49 +55,48 @@ func (e *disableReproduce) apply(_ *ConsumerConfig, prodCfg *ProducerConfig) {
 
 func producerCfg() *ProducerConfig {
 	return &ProducerConfig{
-		EnableReproduce:      DefaultTestProducerConfig.EnableReproduce,
-		CheckPendingInterval: DefaultTestProducerConfig.CheckPendingInterval,
-		KeepAliveTimeout:     DefaultTestProducerConfig.KeepAliveTimeout,
-		CheckResultInterval:  DefaultTestProducerConfig.CheckResultInterval,
+		EnableReproduce:      TestProducerConfig.EnableReproduce,
+		CheckPendingInterval: TestProducerConfig.CheckPendingInterval,
+		KeepAliveTimeout:     TestProducerConfig.KeepAliveTimeout,
+		CheckResultInterval:  TestProducerConfig.CheckResultInterval,
 	}
 }
 
 func consumerCfg() *ConsumerConfig {
 	return &ConsumerConfig{
-		ResponseEntryTimeout: DefaultTestConsumerConfig.ResponseEntryTimeout,
-		KeepAliveTimeout:     DefaultTestConsumerConfig.KeepAliveTimeout,
+		ResponseEntryTimeout: TestConsumerConfig.ResponseEntryTimeout,
+		KeepAliveTimeout:     TestConsumerConfig.KeepAliveTimeout,
 	}
 }
 
 func newProducerConsumers(ctx context.Context, t *testing.T, opts ...configOpt) (*Producer[testRequest, testResponse], []*Consumer[testRequest, testResponse]) {
 	t.Helper()
-	redisURL := redisutil.CreateTestRedis(ctx, t)
+	redisClient, err := redisutil.RedisClientFromURL(redisutil.CreateTestRedis(ctx, t))
+	if err != nil {
+		t.Fatalf("RedisClientFromURL() unexpected error: %v", err)
+	}
 	prodCfg, consCfg := producerCfg(), consumerCfg()
-	prodCfg.RedisURL, consCfg.RedisURL = redisURL, redisURL
-	streamName := uuid.NewString()
-	groupName := fmt.Sprintf("group_%s", streamName)
-	prodCfg.RedisGroup, consCfg.RedisGroup = groupName, groupName
-	prodCfg.RedisStream, consCfg.RedisStream = streamName, streamName
+	streamName := fmt.Sprintf("stream:%s", uuid.NewString())
 	for _, o := range opts {
 		o.apply(consCfg, prodCfg)
 	}
-	producer, err := NewProducer[testRequest, testResponse](prodCfg)
+	producer, err := NewProducer[testRequest, testResponse](redisClient, streamName, prodCfg)
 	if err != nil {
 		t.Fatalf("Error creating new producer: %v", err)
 	}
 
 	var consumers []*Consumer[testRequest, testResponse]
 	for i := 0; i < consumersCount; i++ {
-		c, err := NewConsumer[testRequest, testResponse](ctx, consCfg)
+		c, err := NewConsumer[testRequest, testResponse](redisClient, streamName, consCfg)
 		if err != nil {
 			t.Fatalf("Error creating new consumer: %v", err)
 		}
 		consumers = append(consumers, c)
 	}
-	createGroup(ctx, t, streamName, groupName, producer.client)
+	createRedisGroup(ctx, t, streamName, producer.client)
 	t.Cleanup(func() {
 		ctx := context.Background()
-		destroyGroup(ctx, t, streamName, groupName, producer.client)
+		destroyRedisGroup(ctx, t, streamName, producer.client)
 		var keys []string
 		for _, c := range consumers {
 			keys = append(keys, c.heartBeatKey())