Add flag to configure redis store ttl (flashbots#271)

cmd/geth/main.go

@@ -195,6 +195,7 @@ var (
 		utils.SuaveServiceAlias,
 		utils.SuaveConfidentialTransportRedisEndpointFlag,
 		utils.SuaveConfidentialStoreRedisEndpointFlag,
+		utils.SuaveCondentialStoreRedisTTLFlag,
 		utils.SuaveConfidentialStorePebbleDbPathFlag,
 		utils.SuaveEthBundleSigningKeyFlag,
 		utils.SuaveEthBlockSigningKeyFlag,

cmd/utils/flags.go

@@ -553,6 +553,12 @@ var (
 		Category: flags.SuaveCategory,
 	}
 
+	SuaveCondentialStoreRedisTTLFlag = &cli.DurationFlag{
+		Name:     "suave.confidential.redis-store-ttl",
+		Usage:    "TTL for redis store",
+		Category: flags.SuaveCategory,
+	}
+
 	SuaveConfidentialStorePebbleDbPathFlag = &cli.StringFlag{
 		Name:     "suave.confidential.pebble-store-db-path",
 		Usage:    "Path to pebble db to use for confidential storage backend (default: local store)",
@@ -1756,6 +1762,10 @@ func SetSuaveConfig(ctx *cli.Context, stack *node.Node, cfg *suave.Config) {
 
 	if ctx.IsSet(SuaveConfidentialStoreRedisEndpointFlag.Name) {
 		cfg.RedisStoreUri = ctx.String(SuaveConfidentialStoreRedisEndpointFlag.Name)
+
+		if ctx.IsSet(SuaveCondentialStoreRedisTTLFlag.Name) {
+			cfg.RedisStoreTTL = ctx.Duration(SuaveCondentialStoreRedisTTLFlag.Name)
+		}
 	}
 
 	if ctx.IsSet(SuaveConfidentialStorePebbleDbPathFlag.Name) {

eth/backend.go

@@ -238,7 +238,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 
 	var confidentialStoreBackend cstore.ConfidentialStorageBackend
 	if config.Suave.RedisStoreUri != "" {
-		confidentialStoreBackend, err = cstore.NewRedisStoreBackend(config.Suave.RedisStoreUri)
+		confidentialStoreBackend, err = cstore.NewRedisStoreBackend(config.Suave.RedisStoreUri, config.Suave.RedisStoreTTL)
 		if err != nil {
 			return nil, err
 		}

suave/core/config.go

@@ -1,9 +1,12 @@
 package suave
 
+import "time"
+
 type Config struct {
 	SuaveEthRemoteBackendEndpoint string // deprecated
 	RedisStorePubsubUri           string
 	RedisStoreUri                 string
+	RedisStoreTTL                 time.Duration
 	PebbleDbPath                  string
 	EthBundleSigningKeyHex        string
 	EthBlockSigningKeyHex         string

suave/cstore/pebble_store_backend.go

@@ -94,7 +94,7 @@ func (b *PebbleStoreBackend) InitRecord(record suave.DataRecord) error {
 		if !errors.Is(err, pebble.ErrNotFound) {
 			return err
 		}
-	} else if err == nil {
+	} else {
 		err = json.Unmarshal(rawCurrentValues, &currentValues)
 		closer.Close()
 		if err != nil {

suave/cstore/redis_backends_test.go

@@ -69,14 +69,14 @@ func TestEngineOnRedis(t *testing.T) {
 	mrPubSub := mrStore1
 
 	redisPubSub1 := NewRedisPubSubTransport(mrPubSub.Addr())
-	redisStoreBackend1, _ := NewRedisStoreBackend(mrStore1.Addr())
+	redisStoreBackend1, _ := NewRedisStoreBackend(mrStore1.Addr(), 0)
 
 	engine1 := NewEngine(redisStoreBackend1, redisPubSub1, MockSigner{}, MockChainSigner{})
 	require.NoError(t, engine1.Start())
 	t.Cleanup(func() { engine1.Stop() })
 
 	redisPubSub2 := NewRedisPubSubTransport(mrPubSub.Addr())
-	redisStoreBackend2, _ := NewRedisStoreBackend(mrStore2.Addr())
+	redisStoreBackend2, _ := NewRedisStoreBackend(mrStore2.Addr(), 0)
 
 	engine2 := NewEngine(redisStoreBackend2, redisPubSub2, MockSigner{}, MockChainSigner{})
 	require.NoError(t, engine2.Start())

suave/cstore/redis_store_backend.go

@@ -26,21 +26,26 @@ var (
 		return fmt.Sprintf("record-data-%x-%s", dataId, key)
 	}
 
-	ffStoreTTL = 24 * time.Hour
+	defaultRedisStoreTTL = 24 * time.Hour
 )
 
 type RedisStoreBackend struct {
 	ctx      context.Context
 	cancel   context.CancelFunc
 	redisUri string
+	ttl      time.Duration
 	client   *redis.Client
 	local    *miniredis.Miniredis
 }
 
-func NewRedisStoreBackend(redisUri string) (*RedisStoreBackend, error) {
+func NewRedisStoreBackend(redisUri string, ttl time.Duration) (*RedisStoreBackend, error) {
+	if ttl == 0 {
+		ttl = defaultRedisStoreTTL
+	}
 	r := &RedisStoreBackend{
 		cancel:   nil,
 		redisUri: redisUri,
+		ttl:      ttl,
 	}
 
 	if err := r.start(); err != nil {
@@ -50,6 +55,8 @@ func NewRedisStoreBackend(redisUri string) (*RedisStoreBackend, error) {
 }
 
 func (r *RedisStoreBackend) start() error {
+	log.Info("Starting redis store", "uri", r.redisUri, "ttl", r.ttl)
+
 	if r.redisUri == "" {
 		// create a mini-redis instance
 		localRedis, err := miniredis.Run()
@@ -110,7 +117,7 @@ func (r *RedisStoreBackend) InitRecord(record suave.DataRecord) error {
 		return err
 	}
 
-	err = r.client.Set(r.ctx, key, string(data), ffStoreTTL).Err()
+	err = r.client.Set(r.ctx, key, string(data), r.ttl).Err()
 	if err != nil {
 		return err
 	}
@@ -143,7 +150,7 @@ func (r *RedisStoreBackend) FetchRecordByID(dataId suave.DataId) (suave.DataReco
 
 func (r *RedisStoreBackend) Store(record suave.DataRecord, caller common.Address, key string, value []byte) (suave.DataRecord, error) {
 	storeKey := formatRecordValueKey(record.Id, key)
-	err := r.client.Set(r.ctx, storeKey, string(value), ffStoreTTL).Err()
+	err := r.client.Set(r.ctx, storeKey, string(value), r.ttl).Err()
 	if err != nil {
 		return suave.DataRecord{}, fmt.Errorf("unexpected redis error: %w", err)
 	}

suave/cstore/redis_store_backend_test.go

@@ -2,9 +2,83 @@ package cstore
 
 import (
 	"testing"
+	"time"
+
+	suave "github.com/ethereum/go-ethereum/suave/core"
+	"github.com/stretchr/testify/require"
 )
 
 func TestRedis_StoreSuite(t *testing.T) {
-	store, _ := NewRedisStoreBackend("")
+	store, err := NewRedisStoreBackend("", 0)
+	require.NoError(t, err)
+
 	testBackendStore(t, store)
 }
+
+func TestRedis_TTL_SingleEntry(t *testing.T) {
+	store, err := NewRedisStoreBackend("", 1*time.Second)
+	require.NoError(t, err)
+
+	record1 := suave.DataRecord{
+		Id:                  suave.RandomDataRecordId(),
+		Version:             "a",
+		DecryptionCondition: 1,
+	}
+	require.NoError(t, store.InitRecord(record1))
+
+	record1Found, err := store.FetchRecordByID(record1.Id)
+	require.NoError(t, err)
+	require.Equal(t, record1, record1Found)
+
+	vals := store.FetchRecordsByProtocolAndBlock(1, "a")
+	require.Len(t, vals, 1)
+
+	// Advance past the TTL
+	store.local.FastForward(2 * time.Second)
+
+	_, err = store.FetchRecordByID(record1.Id)
+	require.Error(t, err)
+
+	vals = store.FetchRecordsByProtocolAndBlock(1, "a")
+	require.Len(t, vals, 0)
+}
+
+func TestRedis_TTL_MultipleEntries_SameIndex(t *testing.T) {
+	store, err := NewRedisStoreBackend("", 2*time.Second)
+	require.NoError(t, err)
+
+	record1 := suave.DataRecord{
+		Id:                  suave.RandomDataRecordId(),
+		Version:             "a",
+		DecryptionCondition: 1,
+	}
+	require.NoError(t, store.InitRecord(record1))
+
+	vals := store.FetchRecordsByProtocolAndBlock(1, "a")
+	require.Len(t, vals, 1)
+	require.Equal(t, record1, vals[0])
+
+	// Advance half the TTL time
+	store.local.FastForward(1 * time.Second)
+
+	// Add a new entry that refreshes the index entry
+	record2 := suave.DataRecord{
+		Id:                  suave.RandomDataRecordId(),
+		Version:             "a",
+		DecryptionCondition: 1,
+	}
+	require.NoError(t, store.InitRecord(record2))
+
+	// Advance past the full TTL
+	store.local.FastForward(1 * time.Second)
+
+	_, err = store.FetchRecordByID(record1.Id)
+	require.Error(t, err)
+
+	_, err = store.FetchRecordByID(record2.Id)
+	require.NoError(t, err)
+
+	vals = store.FetchRecordsByProtocolAndBlock(1, "a")
+	require.Len(t, vals, 1)
+	require.Equal(t, record2, vals[0])
+}

suave/cstore/redis_transport.go

@@ -93,10 +93,6 @@ func (r *RedisPubSubTransport) Subscribe() (<-chan DAMessage, context.CancelFunc
 
 			var msg DAMessage
 			msgBytes := common.Hex2Bytes(rmsg.Payload)
-			if err != nil {
-				log.Trace("Redis pubsub: could not decode message from subscription", "err", err, "msg", rmsg.Payload)
-				continue
-			}
 
 			err = json.Unmarshal(msgBytes, &msg)
 			if err != nil {