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

Makefile

@@ -62,17 +62,13 @@ 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
-
 op-devnet-allocs:
 	@echo "Generating devnet allocs..."
 	@./scripts/op-devnet-allocs.sh
 
+benchmark:
+	go test -benchmem -run=^$ -bench . ./e2e -test.parallel 4 -deploy-config ../.devnet/devnetL1.json
+
 .PHONY: \
 	clean \
 	test

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. 
 
@@ -210,6 +214,9 @@ The `raw commitment` is an RLP-encoded [EigenDA certificate](https://github.com/
 
 Unit tests can be ran via invoking `make test`.
 
+### Integration
+End-to-end (E2E) tests can be ran via `make e2e-test`.
+
 ### Holesky
 
 A holesky integration test can be ran using `make holesky-test` to assert proper dispersal/retrieval against a public network. Please **note** that EigenDA Holesky network which is subject to rate-limiting and slow confirmation times *(i.e, >10 minutes per blob confirmation)*. Please advise EigenDA's [inabox](https://github.com/Layr-Labs/eigenda/tree/master/inabox#readme) if you'd like to spin-up a local DA network for faster iteration testing.

cmd/server/entrypoint.go

@@ -29,15 +29,16 @@ 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)
+	sm, err := server.LoadStoreManager(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)
+	server := server.NewServer(cliCtx.String(flags.ListenAddrFlagName), cliCtx.Int(flags.PortFlagName), sm, log, m)
 
 	if err := server.Start(); err != nil {
 		return fmt.Errorf("failed to start the DA server: %w", err)

commitments/mode.go

@@ -7,7 +7,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/benchmark_test.go

@@ -0,0 +1,40 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"testing"
+
+	"github.com/Layr-Labs/eigenda-proxy/client"
+)
+
+// BenchmarkPutsWithSecondary  ... Takes in an async worker count and profiles blob insertions using
+// constant blob sizes in parallel
+func BenchmarkPutsWithSecondary(b *testing.B) {
+	testCfg := TestConfig(true)
+	testCfg.UseS3Caching = true
+	writeThreadCount := os.Getenv("WRITE_THREAD_COUNT")
+	threadInt, err := strconv.Atoi(writeThreadCount)
+	if err != nil {
+		panic(fmt.Errorf("Could not parse WRITE_THREAD_COUNT field %w", err))
+	}
+	testCfg.WriteThreadCount = threadInt
+
+	tsConfig := TestSuiteConfig(testCfg)
+	ts, kill := CreateTestSuite(tsConfig)
+	defer kill()
+
+	cfg := &client.Config{
+		URL: ts.Address(),
+	}
+	daClient := client.New(cfg)
+
+	for i := 0; i < b.N; i++ {
+		_, err := daClient.SetData(context.Background(), []byte("I am a blob and I only live for 14 days on EigenDA"))
+		if err != nil {
+			panic(err)
+		}
+	}
+}

e2e/main_test.go

@@ -1,8 +1,18 @@
 package e2e_test
 
 import (
+	"net/http"
 	"os"
 	"testing"
+
+	"github.com/Layr-Labs/eigenda-proxy/client"
+	"github.com/Layr-Labs/eigenda-proxy/commitments"
+	"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/Layr-Labs/eigenda-proxy/metrics"
+	"github.com/stretchr/testify/require"
 )
 
 // Integration tests are run against memstore whereas.
@@ -11,17 +21,73 @@ import (
 // e.g, in TestProxyServerCaching we only assert to read metrics with EigenDA
 // when referencing memstore since we don't profile the eigenDAClient interactions
 var (
-	runTestnetIntegrationTests bool
-	runIntegrationTests        bool
+	runTestnetIntegrationTests bool // holesky tests
+	runIntegrationTests        bool // memstore tests
 )
 
+// ParseEnv ... reads testing cfg fields. Go test flags don't work for this library due to the dependency on Optimism's E2E framework
+// which initializes test flags per init function which is called before an init in this package.
 func ParseEnv() {
 	runIntegrationTests = os.Getenv("INTEGRATION") == "true" || os.Getenv("INTEGRATION") == "1"
 	runTestnetIntegrationTests = os.Getenv("TESTNET") == "true" || os.Getenv("TESTNET") == "1"
 }
 
+// TestMain ... run main controller
 func TestMain(m *testing.M) {
 	ParseEnv()
 	code := m.Run()
 	os.Exit(code)
 }
+
+// requireDispersalRetrievalEigenDA ... ensure that blob was successfully dispersed/read to/from EigenDA
+func requireDispersalRetrievalEigenDA(t *testing.T, cm *metrics.CountMap, mode commitments.CommitmentMode) {
+	writeCount, err := cm.Get(string(mode), http.MethodPost)
+	require.NoError(t, err)
+	require.True(t, writeCount > 0)
+
+	readCount, err := cm.Get(string(mode), http.MethodGet)
+	require.NoError(t, err)
+	require.True(t, readCount > 0)
+}
+
+// requireWriteReadSecondary ... ensure that secondary backend was successfully written/read to/from
+func requireWriteReadSecondary(t *testing.T, cm *metrics.CountMap, bt store.BackendType) {
+	writeCount, err := cm.Get(http.MethodPut, store.Success, bt.String())
+	require.NoError(t, err)
+	require.True(t, writeCount > 0)
+
+	readCount, err := cm.Get(http.MethodGet, store.Success, bt.String())
+	require.NoError(t, err)
+	require.True(t, readCount > 0)
+}
+
+// requireSimpleClientSetGet ... ensures that simple proxy client can disperse and read a blob
+func requireSimpleClientSetGet(t *testing.T, ts e2e.TestSuite, blob []byte) {
+	cfg := &client.Config{
+		URL: ts.Address(),
+	}
+	daClient := client.New(cfg)
+
+	t.Log("Setting input data on proxy server...")
+	blobInfo, err := daClient.SetData(ts.Ctx, blob)
+	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, blob, preimage)
+
+}
+
+// requireOPClientSetGet ... ensures that alt-da client can disperse and read a blob
+func requireOPClientSetGet(t *testing.T, ts e2e.TestSuite, blob []byte, precompute bool) {
+	daClient := altda.NewDAClient(ts.Address(), false, precompute)
+
+	commit, err := daClient.SetInput(ts.Ctx, blob)
+	require.NoError(t, err)
+
+	preimage, err := daClient.GetInput(ts.Ctx, commit)
+	require.NoError(t, err)
+	require.Equal(t, blob, preimage)
+
+}

e2e/optimism_test.go

@@ -3,6 +3,7 @@ package e2e_test
 import (
 	"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/config"
@@ -125,8 +126,8 @@ func TestOptimismKeccak256Commitment(gt *testing.T) {
 	testCfg := e2e.TestConfig(useMemory())
 	testCfg.UseKeccak256ModeS3 = true
 
-	tsConfig := e2e.TestSuiteConfig(gt, testCfg)
-	proxyTS, shutDown := e2e.CreateTestSuite(gt, tsConfig)
+	tsConfig := e2e.TestSuiteConfig(testCfg)
+	proxyTS, shutDown := e2e.CreateTestSuite(tsConfig)
 	defer shutDown()
 
 	t := actions.NewDefaultTesting(gt)
@@ -167,20 +168,16 @@ 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()
-
-	require.Equal(t, 1, stat.Entries)
-	require.Equal(t, 1, stat.Reads)
+	requireDispersalRetrievalEigenDA(gt, proxyTS.Metrics.HTTPServerRequestsTotal, commitments.OptimismKeccak)
 }
 
 func TestOptimismGenericCommitment(gt *testing.T) {
 	if !runIntegrationTests && !runTestnetIntegrationTests {
 		gt.Skip("Skipping test as INTEGRATION or TESTNET env var not set")
 	}
 
-	tsConfig := e2e.TestSuiteConfig(gt, e2e.TestConfig(useMemory()))
-	proxyTS, shutDown := e2e.CreateTestSuite(gt, tsConfig)
+	tsConfig := e2e.TestSuiteConfig(e2e.TestConfig(useMemory()))
+	proxyTS, shutDown := e2e.CreateTestSuite(tsConfig)
 	defer shutDown()
 
 	t := actions.NewDefaultTesting(gt)
@@ -221,11 +218,5 @@ 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)
-	}
+	requireDispersalRetrievalEigenDA(gt, proxyTS.Metrics.HTTPServerRequestsTotal, commitments.OptimismGeneric)
 }