feat: Optional async decoupling for secondary writes and reworked E2E metric assertions

Makefile

@@ -85,10 +85,6 @@ install-lint:
 	@echo "Installing golangci-lint..."
 	@sh -c $(GET_LINT_CMD)
 
-gosec:
-	@echo "Running security scan with gosec..."
-	gosec ./...
-
 submodules:
 	git submodule update --init --recursive
 

README.md

@@ -65,6 +65,7 @@ In order to disperse to the EigenDA network in production, or at high throughput
 | `--s3.enable-tls` |  | `$EIGENDA_PROXY_S3_ENABLE_TLS` | Enable TLS connection to S3 endpoint. |
 | `--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. |
+| `--routing.concurrent-write-threads` | `0` | `$EIGENDA_PROXY_CONCURRENT_WRITE_THREADS` | Number of threads spun-up for async secondary storage insertions. (<=0) denotes single threaded insertions where (>0) indicates decoupled writes. |
 | `--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 |
@@ -97,6 +98,9 @@ An optional `--eigenda-eth-confirmation-depth` flag can be provided to specify a
 
 An ephemeral memory store backend can be used for faster feedback testing when testing rollup integrations. To target this feature, use the CLI flags `--memstore.enabled`, `--memstore.expiration`.
 
+### Asynchronous Secondary Insertions
+An optional `--routing.concurrent-write-routines` flag can be provided to enable asynchronous processing for secondary writes - allowing for more efficient dispersals in the presence of a hefty secondary routing layer. This flag specifies the number of write routines spun-up with supported thread counts in range `[1, 100)`.
+
 ### Storage Fallback
 An optional storage fallback CLI flag `--routing.fallback-targets` can be leveraged to ensure resiliency when **reading**. When enabled, a blob is persisted to a fallback target after being successfully dispersed. Fallback targets use the keccak256 hash of the existing EigenDA commitment as their key, for succinctness. In the event that blobs cannot be read from EigenDA, they will then be retrieved in linear order from the provided fallback targets. 
 
@@ -208,7 +212,11 @@ The `raw commitment` is an RLP-encoded [EigenDA certificate](https://github.com/
 
 ### Unit
 
-Unit tests can be ran via invoking `make test`.
+Unit tests can be ran via invoking `make test`.  Please make sure to have all test containers downloaded locally before running via:
+```
+docker pull redis
+docker pull minio
+```
 
 ### Holesky
 

cmd/server/entrypoint.go

@@ -29,14 +29,15 @@ func StartProxySvr(cliCtx *cli.Context) error {
 		return fmt.Errorf("failed to pretty print config: %w", err)
 	}
 
+	m := metrics.NewMetrics("default")
+
 	ctx, ctxCancel := context.WithCancel(cliCtx.Context)
 	defer ctxCancel()
 
-	daRouter, err := server.LoadStoreRouter(ctx, cfg, log)
+	daRouter, err := server.LoadStoreRouter(ctx, cfg, log, m)
 	if err != nil {
 		return fmt.Errorf("failed to create store: %w", err)
 	}
-	m := metrics.NewMetrics("default")
 	server := server.NewServer(cliCtx.String(flags.ListenAddrFlagName), cliCtx.Int(flags.PortFlagName), daRouter, log, m)
 
 	if err := server.Start(); err != nil {

commitments/mode.go

@@ -8,7 +8,7 @@ import (
 type CommitmentMeta struct {
 	Mode CommitmentMode
 	// CertVersion is shared for all modes and denotes version of the EigenDA certificate
-	CertVersion byte
+	CertVersion uint8
 }
 
 type CommitmentMode string

e2e/optimism_test.go

@@ -1,8 +1,10 @@
 package e2e_test
 
 import (
+	"net/http"
 	"testing"
 
+	"github.com/Layr-Labs/eigenda-proxy/commitments"
 	"github.com/Layr-Labs/eigenda-proxy/e2e"
 	altda "github.com/ethereum-optimism/optimism/op-alt-da"
 	"github.com/ethereum-optimism/optimism/op-e2e/actions"
@@ -166,11 +168,11 @@ func TestOptimismKeccak256Commitment(gt *testing.T) {
 	optimism.sequencer.ActL2PipelineFull(t)
 	optimism.ActL1Finalized(t)
 
-	// assert that EigenDA proxy's was written and read from
-	stat := proxyTS.Server.GetS3Stats()
+	// assert that EigenDA proxy was written and read from using op keccak256 commitment mode
+	readCount, err := proxyTS.Metrics.HTTPServerRequestsTotal.Find(http.MethodGet, "", string(commitments.OptimismKeccak), "0")
+	require.NoError(t, err)
+	require.True(t, readCount > 0)
 
-	require.Equal(t, 1, stat.Entries)
-	require.Equal(t, 1, stat.Reads)
 }
 
 func TestOptimismGenericCommitment(gt *testing.T) {
@@ -220,11 +222,8 @@ func TestOptimismGenericCommitment(gt *testing.T) {
 	optimism.sequencer.ActL2PipelineFull(t)
 	optimism.ActL1Finalized(t)
 
-	// assert that EigenDA proxy's was written and read from
-
-	if useMemory() {
-		stat := proxyTS.Server.GetEigenDAStats()
-		require.Equal(t, 1, stat.Entries)
-		require.Equal(t, 1, stat.Reads)
-	}
+	// assert that EigenDA proxy was written and read from using op generic commitment mode
+	readCount, err := proxyTS.Metrics.HTTPServerRequestsTotal.Find(http.MethodGet, "", string(commitments.OptimismGeneric), "0")
+	require.NoError(t, err)
+	require.True(t, readCount > 0)
 }

e2e/server_test.go

@@ -8,9 +8,9 @@ import (
 	"time"
 
 	"github.com/Layr-Labs/eigenda-proxy/client"
+	"github.com/Layr-Labs/eigenda-proxy/store"
 
 	"github.com/Layr-Labs/eigenda-proxy/e2e"
-	"github.com/Layr-Labs/eigenda-proxy/store"
 	altda "github.com/ethereum-optimism/optimism/op-alt-da"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -350,9 +350,10 @@ func TestProxyServerCaching(t *testing.T) {
 	require.Equal(t, testPreimage, preimage)
 
 	// ensure that read was from cache
-	s3Stats := ts.Server.GetS3Stats()
-	require.Equal(t, 1, s3Stats.Reads)
-	require.Equal(t, 1, s3Stats.Entries)
+
+	count, err := ts.Metrics.SecondaryRequestsTotal.Find(store.S3BackendType.String(), http.MethodGet, "success")
+	require.NoError(t, err)
+	require.True(t, count > 0)
 
 	if useMemory() { // ensure that eigenda was not read from
 		memStats := ts.Server.GetEigenDAStats()
@@ -393,11 +394,9 @@ func TestProxyServerCachingWithRedis(t *testing.T) {
 	require.Equal(t, testPreimage, preimage)
 
 	// ensure that read was from cache
-	redStats, err := ts.Server.GetStoreStats(store.RedisBackendType)
+	readCount, err := ts.Metrics.SecondaryRequestsTotal.Find(store.RedisBackendType.String(), http.MethodGet, "success")
 	require.NoError(t, err)
-
-	require.Equal(t, 1, redStats.Reads)
-	require.Equal(t, 1, redStats.Entries)
+	require.True(t, readCount > 0)
 
 	if useMemory() { // ensure that eigenda was not read from
 		memStats := ts.Server.GetEigenDAStats()
@@ -420,6 +419,7 @@ func TestProxyServerReadFallback(t *testing.T) {
 
 	t.Parallel()
 
+	// setup server with S3 as a fallback option
 	testCfg := e2e.TestConfig(useMemory())
 	testCfg.UseS3Fallback = true
 	testCfg.Expiration = time.Millisecond * 1
@@ -447,11 +447,11 @@ func TestProxyServerReadFallback(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, testPreimage, preimage)
 
-	// ensure that read was from fallback target location (i.e, S3 for this test)
-	s3Stats := ts.Server.GetS3Stats()
-	require.Equal(t, 1, s3Stats.Reads)
-	require.Equal(t, 1, s3Stats.Entries)
+	count, err := ts.Metrics.SecondaryRequestsTotal.Find(store.S3BackendType.String(), http.MethodGet, "success")
+	require.NoError(t, err)
+	require.True(t, count > 0)
 
+	// TODO - remove this in favor of metrics sampling
 	if useMemory() { // ensure that an eigenda read was attempted with zero data available
 		memStats := ts.Server.GetEigenDAStats()
 		require.Equal(t, 1, memStats.Reads)

e2e/setup.go

@@ -77,7 +77,6 @@ func createS3Config(eigendaCfg server.Config) server.CLIConfig {
 	createS3Bucket(bucketName)
 
 	eigendaCfg.S3Config = s3.Config{
-		Profiling:       true,
 		Bucket:          bucketName,
 		Path:            "",
 		Endpoint:        "localhost:4566",
@@ -175,9 +174,10 @@ func TestSuiteConfig(t *testing.T, testCfg *Cfg) server.CLIConfig {
 }
 
 type TestSuite struct {
-	Ctx    context.Context
-	Log    log.Logger
-	Server *server.Server
+	Ctx     context.Context
+	Log     log.Logger
+	Server  *server.Server
+	Metrics *metrics.InMemoryMetricer
 }
 
 func CreateTestSuite(t *testing.T, testSuiteCfg server.CLIConfig) (TestSuite, func()) {
@@ -187,29 +187,33 @@ func CreateTestSuite(t *testing.T, testSuiteCfg server.CLIConfig) (TestSuite, fu
 		Color:  true,
 	}).New("role", svcName)
 
+	m := metrics.NewInMemoryMetricer()
 	ctx := context.Background()
 	store, err := server.LoadStoreRouter(
 		ctx,
 		testSuiteCfg,
 		log,
+		m,
 	)
+
 	require.NoError(t, err)
-	server := server.NewServer(host, 0, store, log, metrics.NoopMetrics)
+	proxySvr := server.NewServer(host, 0, store, log, m)
 
 	t.Log("Starting proxy server...")
-	err = server.Start()
+	err = proxySvr.Start()
 	require.NoError(t, err)
 
 	kill := func() {
-		if err := server.Stop(); err != nil {
-			panic(err)
+		if err := proxySvr.Stop(); err != nil {
+			log.Error("failed to stop proxy server", "err", err)
 		}
 	}
 
 	return TestSuite{
-		Ctx:    ctx,
-		Log:    log,
-		Server: server,
+		Ctx:     ctx,
+		Log:     log,
+		Server:  proxySvr,
+		Metrics: m,
 	}, kill
 }
 

flags/flags.go

@@ -28,8 +28,10 @@ const (
 	PortFlagName       = "port"
 
 	// routing flags
+	// TODO: change "routing" --> "secondary"
 	FallbackTargetsFlagName = "routing.fallback-targets"
 	CacheTargetsFlagName    = "routing.cache-targets"
+	ConcurrentWriteThreads  = "routing.concurrent-write-routines"
 )
 
 const EnvVarPrefix = "EIGENDA_PROXY"
@@ -43,13 +45,13 @@ func CLIFlags() []cli.Flag {
 	flags := []cli.Flag{
 		&cli.StringFlag{
 			Name:    ListenAddrFlagName,
-			Usage:   "server listening address",
+			Usage:   "Server listening address",
 			Value:   "0.0.0.0",
 			EnvVars: prefixEnvVars("ADDR"),
 		},
 		&cli.IntFlag{
 			Name:    PortFlagName,
-			Usage:   "server listening port",
+			Usage:   "Server listening port",
 			Value:   3100,
 			EnvVars: prefixEnvVars("PORT"),
 		},
@@ -65,6 +67,12 @@ func CLIFlags() []cli.Flag {
 			Value:   cli.NewStringSlice(),
 			EnvVars: prefixEnvVars("CACHE_TARGETS"),
 		},
+		&cli.IntFlag{
+			Name:    ConcurrentWriteThreads,
+			Usage:   "Number of threads spun-up for async secondary storage insertions. (<=0) denotes single threaded insertions where (>0) indicates decoupled writes.",
+			Value:   0,
+			EnvVars: prefixEnvVars("CONCURRENT_WRITE_THREADS"),
+		},
 	}
 
 	return flags

metrics/memory.go

@@ -0,0 +1,119 @@
+package metrics
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+
+	"github.com/ethereum-optimism/optimism/op-service/metrics"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+func keyLabels(labels []string) (common.Hash, error) {
+	sort.Strings(labels) // in-place sort strings so keys are order agnostic
+
+	encodedBytes, err := rlp.EncodeToBytes(labels)
+	if err != nil {
+		return common.Hash{}, err
+	}
+
+	hash := crypto.Keccak256Hash(encodedBytes)
+
+	return hash, nil
+}
+
+type MetricCountMap struct {
+	m *sync.Map
+}
+
+func NewCountMap() *MetricCountMap {
+	return &MetricCountMap{
+		m: new(sync.Map),
+	}
+}
+
+func (mcm *MetricCountMap) insert(values ...string) error {
+	key, err := keyLabels(values)
+
+	if err != nil {
+		return err
+	}
+
+	// update or add count entry
+	value, exists := mcm.m.Load(key.Hex())
+	if !exists {
+		mcm.m.Store(key.Hex(), uint64(1))
+		return nil
+	}
+	uint64Val, ok := value.(uint64)
+	if !ok {
+		return fmt.Errorf("could not read uint64 from sync map")
+	}
+
+	mcm.m.Store(key.Hex(), uint64Val+uint64(1))
+	return nil
+}
+
+func (mcm *MetricCountMap) Find(values ...string) (uint64, error) {
+	key, err := keyLabels(values)
+
+	if err != nil {
+		return 0, err
+	}
+
+	val, exists := mcm.m.Load(key.Hex())
+	if !exists {
+		return 0, fmt.Errorf("value doesn't exist")
+	}
+	uint64Val, ok := val.(uint64)
+	if !ok {
+		return 0, fmt.Errorf("could not read uint64 from sync map")
+	}
+
+	return uint64Val, nil
+}
+
+type InMemoryMetricer struct {
+	HTTPServerRequestsTotal *MetricCountMap
+	// secondary metrics
+	SecondaryRequestsTotal *MetricCountMap
+}
+
+func NewInMemoryMetricer() *InMemoryMetricer {
+	return &InMemoryMetricer{
+		HTTPServerRequestsTotal: NewCountMap(),
+		SecondaryRequestsTotal:  NewCountMap(),
+	}
+}
+
+var _ Metricer = NewInMemoryMetricer()
+
+func (n *InMemoryMetricer) Document() []metrics.DocumentedMetric {
+	return nil
+}
+
+func (n *InMemoryMetricer) RecordInfo(_ string) {
+}
+
+func (n *InMemoryMetricer) RecordUp() {
+}
+
+func (n *InMemoryMetricer) RecordRPCServerRequest(endpoint string) func(status, mode, ver string) {
+	return func(x string, y string, z string) {
+		err := n.HTTPServerRequestsTotal.insert(endpoint, x, y, z)
+		if err != nil {
+			panic(err)
+		}
+	}
+}
+
+func (n *InMemoryMetricer) RecordSecondaryRequest(x string, y string) func(status string) {
+	return func(z string) {
+		err := n.SecondaryRequestsTotal.insert(x, y, z)
+		if err != nil {
+			panic(err)
+		}
+	}
+}