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

Makefile

@@ -62,13 +62,13 @@ install-lint:
 	@echo "Installing golangci-lint..."
 	@sh -c $(GET_LINT_CMD)
 
-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

cmd/server/entrypoint.go

@@ -34,11 +34,11 @@ func StartProxySvr(cliCtx *cli.Context) error {
 	ctx, ctxCancel := context.WithCancel(cliCtx.Context)
 	defer ctxCancel()
 
-	daRouter, err := server.LoadStoreRouter(ctx, cfg, log, m)
+	sm, err := server.LoadStoreManager(ctx, cfg, log, m)
 	if err != nil {
 		return fmt.Errorf("failed to create store: %w", err)
 	}
-	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)

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

@@ -1,7 +1,6 @@
 package e2e_test
 
 import (
-	"net/http"
 	"testing"
 
 	"github.com/Layr-Labs/eigenda-proxy/commitments"
@@ -127,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)
@@ -169,20 +168,16 @@ func TestOptimismKeccak256Commitment(gt *testing.T) {
 	optimism.sequencer.ActL2PipelineFull(t)
 	optimism.ActL1Finalized(t)
 
-	// 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)
-
+	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)
@@ -223,8 +218,5 @@ func TestOptimismGenericCommitment(gt *testing.T) {
 	optimism.sequencer.ActL2PipelineFull(t)
 	optimism.ActL1Finalized(t)
 
-	// 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)
+	requireDispersalRetrievalEigenDA(gt, proxyTS.Metrics.HTTPServerRequestsTotal, commitments.OptimismGeneric)
 }