feat: Redis backend target (#111)

* feat: Redis backend target

* feat: Redis backend target - update holesky test make target

* feat: Redis backend target - address PR feedback

* feat: Redis backend target - rebase to new main

Makefile

@@ -26,11 +26,19 @@ docker-build:
 run-minio:
  docker run -p 4566:9000 -d -e "MINIO_ROOT_USER=minioadmin" -e "MINIO_ROOT_PASSWORD=minioadmin" --name minio minio/minio server /data
 
+run-redis:
+ docker run -p 9001:6379 -d --name redis redis
+
 stop-minio:
  @if [ -n "$$(docker ps -q -f name=minio)" ]; then \
  docker stop minio && docker rm minio; \
  fi
 
+stop-redis:
+ @if [ -n "$$(docker ps -q -f name=redis)" ]; then \
+ docker stop redis && docker rm redis; \
+ fi
+
 run-memstore-server:
  ./bin/eigenda-proxy --memstore.enabled
 
@@ -40,13 +48,15 @@ clean:
 test:
  go test -v ./... -parallel 4 
 
-e2e-test: stop-minio run-minio
+e2e-test: stop-minio stop-redis run-minio run-redis
  $(E2ETEST) && \
- make stop-minio
+ make stop-minio && \
+ make stop-redis
 
-holesky-test: run-minio
+holesky-test: stop-minio stop-redis run-minio run-redis
  $(HOLESKYTEST) && \
- make stop-minio
+ make stop-minio && \
+ make stop-redis
 
 .PHONY: lint
 lint:

README.md

@@ -64,6 +64,10 @@ In order to disperse to the EigenDA network in production, or at high throughput
 | `--routing.fallback-targets` | `[]` | `$EIGENDA_PROXY_FALLBACK_TARGETS` | Fall back backend targets. Supports S3. | Backup storage locations to read from in the event of eigenda retrieval failure. |
 | `--routing.cache-targets` | `[]` | `$EIGENDA_PROXY_CACHE_TARGETS` | Caching targets. Supports S3. | Caches data to backend targets after dispersing to DA, retrieved from before trying read from EigenDA. |
 | `--s3.timeout` | `5s` | `$EIGENDA_PROXY_S3_TIMEOUT` | timeout for S3 storage operations (e.g. get, put) |
+| `--redis.db` | `0` | `$EIGENDA_PROXY_REDIS_DB` | redis database to use after connecting to server |
+| `--redis.endpoint` | `""` | `$EIGENDA_PROXY_REDIS_ENDPOINT` | redis endpoint url |
+| `--redis.password` | `""` | `$EIGENDA_PROXY_REDIS_PASSWORD` | redis password |
+| `--redis.eviction` | `24h0m0s` | `$EIGENDA_PROXY_REDIS_EVICTION` | entry eviction/expiration time |
 | `--help, -h` | `false` | | Show help. |
 | `--version, -v` | `false` | | Print the version. |
 

e2e/server_test.go

@@ -7,6 +7,7 @@ import (
 
  "github.com/Layr-Labs/eigenda-proxy/client"
  "github.com/Layr-Labs/eigenda-proxy/e2e"
+ "github.com/Layr-Labs/eigenda-proxy/store"
  op_plasma "github.com/ethereum-optimism/optimism/op-plasma"
  "github.com/stretchr/testify/require"
 )
@@ -203,6 +204,50 @@ func TestProxyServerCaching(t *testing.T) {
  }
 }
 
+func TestProxyServerCachingWithRedis(t *testing.T) {
+ if !runIntegrationTests && !runTestnetIntegrationTests {
+ t.Skip("Skipping test as INTEGRATION or TESTNET env var not set")
+ }
+
+ t.Parallel()
+
+ testCfg := e2e.TestConfig(useMemory())
+ testCfg.UseRedisCaching = true
+
+ ts, kill := e2e.CreateTestSuite(t, testCfg)
+ defer kill()
+
+ cfg := &client.Config{
+ URL: ts.Address(),
+ }
+ daClient := client.New(cfg)
+ // 10 kb blob
+ testPreimage := []byte(e2e.RandString(10_000))
+
+ t.Log("Setting input data on proxy server...")
+ blobInfo, err := daClient.SetData(ts.Ctx, testPreimage)
+ require.NotEmpty(t, blobInfo)
+ require.NoError(t, err)
+
+ t.Log("Getting input data from proxy server...")
+ preimage, err := daClient.GetData(ts.Ctx, blobInfo)
+ require.NoError(t, err)
+ require.Equal(t, testPreimage, preimage)
+
+ // ensure that read was from cache
+ redStats, err := ts.Server.GetStoreStats(store.Redis)
+ require.NoError(t, err)
+
+ require.Equal(t, 1, redStats.Reads)
+ require.Equal(t, 1, redStats.Entries)
+
+ if useMemory() { // ensure that eigenda was not read from
+ memStats := ts.Server.GetEigenDAStats()
+ require.Equal(t, 0, memStats.Reads)
+ require.Equal(t, 1, memStats.Entries)
+ }
+}
+
 /*
  Ensure that fallback location is read from when EigenDA blob is not available.
  This is done by setting the memstore expiration time to 1ms and waiting for the blob to expire

e2e/setup.go

@@ -36,6 +36,7 @@ type Cfg struct {
  Expiration time.Duration
  UseKeccak256ModeS3 bool
  UseS3Caching bool
+ UseRedisCaching bool
  UseS3Fallback bool
 }
 
@@ -45,10 +46,24 @@ func TestConfig(useMemory bool) *Cfg {
  Expiration: 14 * 24 * time.Hour,
  UseKeccak256ModeS3: false,
  UseS3Caching: false,
+ UseRedisCaching: false,
  UseS3Fallback: false,
  }
 }
 
+func createRedisConfig(eigendaCfg server.Config) server.CLIConfig {
+ return server.CLIConfig{
+ EigenDAConfig: eigendaCfg,
+ RedisCfg: store.RedisConfig{
+ Endpoint: "127.0.0.1:9001",
+ Password: "",
+ DB: 0,
+ Eviction: 10 * time.Minute,
+ Profile: true,
+ },
+ }
+}
+
 func createS3Config(eigendaCfg server.Config) server.CLIConfig {
  // generate random string
  bucketName := "eigenda-proxy-test-" + RandString(10)
@@ -141,6 +156,10 @@ func CreateTestSuite(t *testing.T, testCfg *Cfg) (TestSuite, func()) {
  eigendaCfg.FallbackTargets = []string{"S3"}
  cfg = createS3Config(eigendaCfg)
 
+ case testCfg.UseRedisCaching:
+ eigendaCfg.CacheTargets = []string{"redis"}
+ cfg = createRedisConfig(eigendaCfg)
+
  default:
  cfg = server.CLIConfig{
  EigenDAConfig: eigendaCfg,

go.mod

@@ -66,6 +66,7 @@ require (
  github.com/decred/dcrd/crypto/blake256 v1.0.1 // indirect
  github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
  github.com/deepmap/oapi-codegen v1.8.2 // indirect
+ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
  github.com/dlclark/regexp2 v1.7.0 // indirect
  github.com/docker/go-units v0.5.0 // indirect
  github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127 // indirect
@@ -88,6 +89,7 @@ require (
  github.com/getsentry/sentry-go v0.20.0 // indirect
  github.com/go-ini/ini v1.67.0 // indirect
  github.com/go-ole/go-ole v1.3.0 // indirect
+ github.com/go-redis/redis/v8 v8.11.5 // indirect
  github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect
  github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
  github.com/goccy/go-json v0.10.3 // indirect

go.sum

@@ -226,6 +226,8 @@ github.com/dgraph-io/badger v1.6.2/go.mod h1:JW2yswe3V058sS0kZ2h/AXeDSqFjxnZcRrV
 github.com/dgraph-io/ristretto v0.0.3-0.20200630154024-f66de99634de h1:t0UHb5vdojIDUqktM6+xJAfScFBsVpXZmqC9dsgJmeA=
 github.com/dgraph-io/ristretto v0.0.3-0.20200630154024-f66de99634de/go.mod h1:KPxhHT9ZxKefz+PCeOGsrHpl1qZ7i70dGTu2u+Ahh6E=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/distribution/reference v0.5.0 h1:/FUIFXtfc/x2gpa5/VGfiGLuOIdYa1t65IKK2OFGvA0=
 github.com/distribution/reference v0.5.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
 github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
@@ -315,6 +317,8 @@ github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
 github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
 github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
+github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
+github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
 github.com/go-sourcemap/sourcemap v2.1.3+incompatible h1:W1iEw64niKVGogNgBN3ePyLFfuisuzeidWPMPWmECqU=
 github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=

server/config.go

@@ -41,6 +41,12 @@ const (
  MemstorePutLatencyFlagName = "memstore.put-latency"
  MemstoreGetLatencyFlagName = "memstore.get-latency"
 
+ // redis client flags
+ RedisEndpointFlagName = "redis.endpoint"
+ RedisPasswordFlagName = "redis.password"
+ RedisDBFlagName = "redis.db"
+ RedisEvictionFlagName = "redis.eviction"
+
  // S3 client flags
  S3CredentialTypeFlagName = "s3.credential-type" // #nosec G101
  S3BucketFlagName = "s3.bucket" // #nosec G101
@@ -67,29 +73,28 @@ var (
 )
 
 type Config struct {
- S3Config store.S3Config
-
+ // eigenda
  ClientConfig clients.EigenDAClientConfig
 
- // The blob encoding version to use when writing blobs from the high level interface.
+ // the blob encoding version to use when writing blobs from the high level interface.
  PutBlobEncodingVersion codecs.BlobEncodingVersion
 
- // ETH vars
+ // eth vars
  EthRPC string
  SvcManagerAddr string
  EthConfirmationDepth int64
 
- // KZG vars
+ // kzg vars
  CacheDir string
  G1Path string
  G2Path string
  G2PowerOfTauPath string
 
- // Size constraints
+ // size constraints
  MaxBlobLength string
  maxBlobLengthBytes uint64
 
- // Memstore
+ // memstore
  MemstoreEnabled bool
  MemstoreBlobExpiration time.Duration
  MemstoreGetLatency time.Duration
@@ -98,6 +103,10 @@ type Config struct {
  // routing
  FallbackTargets []string
  CacheTargets []string
+
+ // secondary storage
+ RedisCfg store.RedisConfig
+ S3Config store.S3Config
 }
 
 // GetMaxBlobLength ... returns the maximum blob length in bytes
@@ -153,6 +162,12 @@ func (cfg *Config) VerificationCfg() *verify.Config {
 // ReadConfig ... parses the Config from the provided flags or environment variables.
 func ReadConfig(ctx *cli.Context) Config {
  cfg := Config{
+ RedisCfg: store.RedisConfig{
+ Endpoint: ctx.String(RedisEndpointFlagName),
+ Password: ctx.String(RedisPasswordFlagName),
+ DB: ctx.Int(RedisDBFlagName),
+ Eviction: ctx.Duration(RedisEvictionFlagName),
+ },
  S3Config: store.S3Config{
  S3CredentialType: store.StringToS3CredentialType(ctx.String(S3CredentialTypeFlagName)),
  Bucket: ctx.String(S3BucketFlagName),
@@ -244,6 +259,10 @@ func (cfg *Config) Check() error {
  }
  }
 
+ if cfg.RedisCfg.Endpoint == "" && cfg.RedisCfg.Password != "" {
+ return fmt.Errorf("redis password is set, but endpoint is not")
+ }
+
  if !cfg.MemstoreEnabled && cfg.ClientConfig.RPC == "" {
  return fmt.Errorf("eigenda disperser rpc url is not set")
  }
@@ -268,7 +287,7 @@ func (cfg *Config) Check() error {
  return nil
 }
 
-// flags used for S3 backend configuration
+// s3Flags ... used for S3 backend configuration
 func s3Flags() []cli.Flag {
  return []cli.Flag{
  &cli.StringFlag{
@@ -319,6 +338,34 @@ func s3Flags() []cli.Flag {
  }
 }
 
+// redisFlags ... used for Redis backend configuration
+func redisFlags() []cli.Flag {
+ return []cli.Flag{
+ &cli.StringFlag{
+ Name: RedisEndpointFlagName,
+ Usage: "Redis endpoint",
+ EnvVars: prefixEnvVars("REDIS_ENDPOINT"),
+ },
+ &cli.StringFlag{
+ Name: RedisPasswordFlagName,
+ Usage: "Redis password",
+ EnvVars: prefixEnvVars("REDIS_PASSWORD"),
+ },
+ &cli.IntFlag{
+ Name: RedisDBFlagName,
+ Usage: "Redis database",
+ Value: 0,
+ EnvVars: prefixEnvVars("REDIS_DB"),
+ },
+ &cli.DurationFlag{
+ Name: RedisEvictionFlagName,
+ Usage: "Redis eviction time",
+ Value: 24 * time.Hour,
+ EnvVars: prefixEnvVars("REDIS_EVICTION"),
+ },
+ }
+}
+
 func CLIFlags() []cli.Flag {
  // TODO: Decompose all flags into constituent parts based on their respective category / usage
  flags := []cli.Flag{
@@ -452,5 +499,6 @@ func CLIFlags() []cli.Flag {
  }
 
  flags = append(flags, s3Flags()...)
+ flags = append(flags, redisFlags()...)
  return flags
 }

server/config_test.go

@@ -11,6 +11,12 @@ import (
 
 func validCfg() *Config {
  return &Config{
+ RedisCfg: store.RedisConfig{
+ Endpoint: "localhost:6379",
+ Password: "password",
+ DB: 0,
+ Eviction: 10 * time.Minute,
+ },
  S3Config: store.S3Config{
  Bucket: "test-bucket",
  Path: "",
@@ -174,4 +180,12 @@ func TestConfigVerification(t *testing.T) {
  err := cfg.Check()
  require.Error(t, err)
  })
+
+ t.Run("BadRedisConfiguration", func(t *testing.T) {
+ cfg := validCfg()
+ cfg.RedisCfg.Endpoint = ""
+
+ err := cfg.Check()
+ require.Error(t, err)
+ })
 }

server/flags.go

@@ -43,6 +43,7 @@ func init() {
 }
 
 type CLIConfig struct {
+ RedisCfg store.RedisConfig
  S3Config store.S3Config
  EigenDAConfig Config
  MetricsCfg opmetrics.CLIConfig
@@ -51,6 +52,7 @@ type CLIConfig struct {
 func ReadCLIConfig(ctx *cli.Context) CLIConfig {
  config := ReadConfig(ctx)
  return CLIConfig{
+ RedisCfg: config.RedisCfg,
  EigenDAConfig: config,
  MetricsCfg: opmetrics.ReadCLIConfig(ctx),
  S3Config: config.S3Config,