Merge mempool struct to confstore engine (#51)

* Merge mempool struct to confstore engine

* Remove legacy interface

* Revert "Remove legacy interface"

This reverts commit b921ff8.

* Revert "Merge mempool struct to confstore engine"

This reverts commit be07a6f.

* Merge mempool in store

* Remove mempool impl

core/vm/contracts_suave_test.go

@@ -77,7 +77,7 @@ func TestSuavePrecompileStub(t *testing.T) {
 	// This test ensures that the Suave precompile stubs work as expected
 	// for encoding/decoding.
 	mockSuaveBackend := &mockSuaveBackend{}
-	stubEngine, err := suave.NewConfidentialStoreEngine(mockSuaveBackend, mockSuaveBackend, suave.MockMempool{}, suave.MockSigner{}, suave.MockChainSigner{})
+	stubEngine, err := suave.NewConfidentialStoreEngine(mockSuaveBackend, mockSuaveBackend, suave.MockSigner{}, suave.MockChainSigner{})
 	require.NoError(t, err)
 
 	suaveContext := SuaveContext{
@@ -141,8 +141,7 @@ func TestSuavePrecompileStub(t *testing.T) {
 
 func newTestBackend(t *testing.T) *suaveRuntime {
 	confStore := backends.NewLocalConfidentialStore()
-	suaveMempool := backends.NewMempoolOnConfidentialStore(confStore)
-	confEngine, err := suave.NewConfidentialStoreEngine(confStore, &suave.MockTransport{}, suaveMempool, suave.MockSigner{}, suave.MockChainSigner{})
+	confEngine, err := suave.NewConfidentialStoreEngine(confStore, &suave.MockTransport{}, suave.MockSigner{}, suave.MockChainSigner{})
 	require.NoError(t, err)
 
 	require.NoError(t, confEngine.Start())

eth/backend.go

@@ -248,10 +248,9 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		suaveEthBackend = &suave_backends.EthMock{}
 	}
 
-	suaveBidMempool := suave_backends.NewMempoolOnConfidentialStore(confidentialStoreBackend)
 	suaveDaSigner := &suave_backends.AccountManagerDASigner{Manager: eth.AccountManager()}
 
-	confidentialStoreEngine, err := suave.NewConfidentialStoreEngine(confidentialStoreBackend, confidentialStoreTransport, suaveBidMempool, suaveDaSigner, types.LatestSigner(chainConfig))
+	confidentialStoreEngine, err := suave.NewConfidentialStoreEngine(confidentialStoreBackend, confidentialStoreTransport, suaveDaSigner, types.LatestSigner(chainConfig))
 	if err != nil {
 		return nil, err
 	}

suave/backends/redis_backends_test.go

@@ -97,7 +97,7 @@ func TestEngineOnRedis(t *testing.T) {
 	redisPubSub1 := NewRedisPubSubTransport(mrPubSub.Addr())
 	redisStoreBackend1 := NewRedisStoreBackend(mrStore1.Addr())
 
-	engine1, err := suave.NewConfidentialStoreEngine(redisStoreBackend1, redisPubSub1, suave.MockMempool{}, suave.MockSigner{}, suave.MockChainSigner{})
+	engine1, err := suave.NewConfidentialStoreEngine(redisStoreBackend1, redisPubSub1, suave.MockSigner{}, suave.MockChainSigner{})
 	require.NoError(t, err)
 
 	require.NoError(t, engine1.Start())
@@ -106,7 +106,7 @@ func TestEngineOnRedis(t *testing.T) {
 	redisPubSub2 := NewRedisPubSubTransport(mrPubSub.Addr())
 	redisStoreBackend2 := NewRedisStoreBackend(mrStore2.Addr())
 
-	engine2, err := suave.NewConfidentialStoreEngine(redisStoreBackend2, redisPubSub2, suave.MockMempool{}, suave.MockSigner{}, suave.MockChainSigner{})
+	engine2, err := suave.NewConfidentialStoreEngine(redisStoreBackend2, redisPubSub2, suave.MockSigner{}, suave.MockChainSigner{})
 	require.NoError(t, err)
 
 	require.NoError(t, engine2.Start())

suave/backends/redis_store_backend.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/alicebob/miniredis/v2"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
 	suave "github.com/ethereum/go-ethereum/suave/core"
 	"github.com/go-redis/redis/v8"
 )
@@ -38,10 +40,11 @@ func NewLocalConfidentialStore() *RedisStoreBackend {
 }
 
 func NewRedisStoreBackend(redisUri string) *RedisStoreBackend {
-	return &RedisStoreBackend{
+	r := &RedisStoreBackend{
 		cancel:   nil,
 		redisUri: redisUri,
 	}
+	return r
 }
 
 func (r *RedisStoreBackend) Start() error {
@@ -69,6 +72,11 @@ func (r *RedisStoreBackend) Start() error {
 	}
 	r.client = client
 
+	err = r.InitializeBid(mempoolConfidentialStoreBid)
+	if err != nil && !errors.Is(err, suave.ErrBidAlreadyPresent) {
+		return fmt.Errorf("mempool: could not initialize: %w", err)
+	}
+
 	return nil
 }
 
@@ -143,3 +151,51 @@ func (r *RedisStoreBackend) Retrieve(bid suave.Bid, caller common.Address, key s
 
 	return data, nil
 }
+
+var (
+	mempoolConfStoreId          = types.BidId{0x39}
+	mempoolConfStoreAddr        = common.HexToAddress("0x39")
+	mempoolConfidentialStoreBid = suave.Bid{Id: mempoolConfStoreId, AllowedPeekers: []common.Address{mempoolConfStoreAddr}}
+)
+
+func (r *RedisStoreBackend) SubmitBid(bid types.Bid) error {
+	defer log.Info("bid submitted", "bid", bid, "store", r.Store)
+
+	var bidsByBlockAndProtocol []types.Bid
+	bidsByBlockAndProtocolBytes, err := r.Retrieve(mempoolConfidentialStoreBid, mempoolConfStoreAddr, fmt.Sprintf("protocol-%s-bn-%d", bid.Version, bid.DecryptionCondition))
+	if err == nil {
+		bidsByBlockAndProtocol = suave.MustDecode[[]types.Bid](bidsByBlockAndProtocolBytes)
+	}
+	// store bid by block number and by protocol + block number
+	bidsByBlockAndProtocol = append(bidsByBlockAndProtocol, bid)
+
+	r.Store(mempoolConfidentialStoreBid, mempoolConfStoreAddr, fmt.Sprintf("protocol-%s-bn-%d", bid.Version, bid.DecryptionCondition), suave.MustEncode(bidsByBlockAndProtocol))
+
+	return nil
+}
+
+func (r *RedisStoreBackend) FetchBidById(bidId suave.BidId) (types.Bid, error) {
+	engineBid, err := r.FetchEngineBidById(bidId)
+	if err != nil {
+		log.Error("bid missing!", "id", bidId, "err", err)
+		return types.Bid{}, errors.New("not found")
+	}
+
+	return types.Bid{
+		Id:                  engineBid.Id,
+		Salt:                engineBid.Salt,
+		DecryptionCondition: engineBid.DecryptionCondition,
+		AllowedPeekers:      engineBid.AllowedPeekers,
+		AllowedStores:       engineBid.AllowedStores,
+		Version:             engineBid.Version,
+	}, nil
+}
+
+func (r *RedisStoreBackend) FetchBidsByProtocolAndBlock(blockNumber uint64, namespace string) []types.Bid {
+	bidsByProtocolBytes, err := r.Retrieve(mempoolConfidentialStoreBid, mempoolConfStoreAddr, fmt.Sprintf("protocol-%s-bn-%d", namespace, blockNumber))
+	if err != nil {
+		return nil
+	}
+	defer log.Info("bids fetched", "bids", string(bidsByProtocolBytes))
+	return suave.MustDecode[[]types.Bid](bidsByProtocolBytes)
+}

suave/core/engine.go

@@ -19,7 +19,6 @@ type ConfidentialStoreEngine struct {
 
 	backend        ConfidentialStoreBackend
 	transportTopic StoreTransportTopic
-	mempool        MempoolBackend
 
 	daSigner    DASigner
 	chainSigner ChainSigner
@@ -37,10 +36,6 @@ func (e *ConfidentialStoreEngine) Start() error {
 		return err
 	}
 
-	if err := e.mempool.Start(); err != nil {
-		return err
-	}
-
 	if e.cancel != nil {
 		e.cancel()
 	}
@@ -59,9 +54,6 @@ func (e *ConfidentialStoreEngine) Stop() error {
 	}
 
 	e.cancel()
-	if err := e.mempool.Stop(); err != nil {
-		log.Warn("Confidential engine: error while stopping mempool", "err", err)
-	}
 
 	if err := e.transportTopic.Stop(); err != nil {
 		log.Warn("Confidential engine: error while stopping transport", "err", err)
@@ -84,7 +76,7 @@ type ChainSigner interface {
 	Sender(tx *types.Transaction) (common.Address, error)
 }
 
-func NewConfidentialStoreEngine(backend ConfidentialStoreBackend, transportTopic StoreTransportTopic, mempool MempoolBackend, daSigner DASigner, chainSigner ChainSigner) (*ConfidentialStoreEngine, error) {
+func NewConfidentialStoreEngine(backend ConfidentialStoreBackend, transportTopic StoreTransportTopic, daSigner DASigner, chainSigner ChainSigner) (*ConfidentialStoreEngine, error) {
 	localAddresses := make(map[common.Address]struct{})
 	for _, addr := range daSigner.LocalAddresses() {
 		localAddresses[addr] = struct{}{}
@@ -93,7 +85,6 @@ func NewConfidentialStoreEngine(backend ConfidentialStoreBackend, transportTopic
 	engine := &ConfidentialStoreEngine{
 		backend:        backend,
 		transportTopic: transportTopic,
-		mempool:        mempool,
 		daSigner:       daSigner,
 		chainSigner:    chainSigner,
 		storeUUID:      uuid.New(),
@@ -180,23 +171,23 @@ func (e *ConfidentialStoreEngine) InitializeBid(bid types.Bid, creationTx *types
 	}
 
 	// send the bid to the internal mempool
-	if err := e.mempool.SubmitBid(bid); err != nil {
+	if err := e.backend.SubmitBid(bid); err != nil {
 		return types.Bid{}, fmt.Errorf("failed to submit to mempool: %w", err)
 	}
 
 	return bid, nil
 }
 
 func (e *ConfidentialStoreEngine) SubmitBid(bid types.Bid) error {
-	return e.mempool.SubmitBid(bid)
+	return e.backend.SubmitBid(bid)
 }
 
 func (e *ConfidentialStoreEngine) FetchBidById(bidId BidId) (types.Bid, error) {
-	return e.mempool.FetchBidById(bidId)
+	return e.backend.FetchBidById(bidId)
 }
 
 func (e *ConfidentialStoreEngine) FetchBidsByProtocolAndBlock(blockNumber uint64, namespace string) []types.Bid {
-	return e.mempool.FetchBidsByProtocolAndBlock(blockNumber, namespace)
+	return e.backend.FetchBidsByProtocolAndBlock(blockNumber, namespace)
 }
 
 func (e *ConfidentialStoreEngine) Store(bidId BidId, sourceTx *types.Transaction, caller common.Address, key string, value []byte) (Bid, error) {
@@ -339,7 +330,7 @@ func (e *ConfidentialStoreEngine) NewMessage(message DAMessage) error {
 			return fmt.Errorf("unexpected error while initializing bid from transport: %w", err)
 		}
 	} else {
-		e.mempool.SubmitBid(innerBid)
+		e.backend.SubmitBid(innerBid)
 	}
 
 	_, err = e.backend.Store(message.Bid, message.Caller, message.Key, message.Value)
@@ -405,15 +396,3 @@ func (MockChainSigner) Sender(tx *types.Transaction) (common.Address, error) {
 
 	return *tx.To(), nil
 }
-
-type MockMempool struct{}
-
-func (MockMempool) Start() error { return nil }
-func (MockMempool) Stop() error  { return nil }
-
-func (MockMempool) SubmitBid(types.Bid) error { return nil }
-
-func (MockMempool) FetchBidById(BidId) (types.Bid, error) { return types.Bid{}, nil }
-func (MockMempool) FetchBidsByProtocolAndBlock(blockNumber uint64, namespace string) []types.Bid {
-	return nil
-}

suave/core/engine_test.go

@@ -42,6 +42,18 @@ func (*FakeStoreBackend) Retrieve(bid Bid, caller common.Address, key string) ([
 	return nil, errors.New("not implemented")
 }
 
+func (*FakeStoreBackend) FetchBidById(BidId) (types.Bid, error) {
+	return types.Bid{}, nil
+}
+
+func (*FakeStoreBackend) FetchBidsByProtocolAndBlock(blockNumber uint64, namespace string) []types.Bid {
+	return nil
+}
+
+func (*FakeStoreBackend) SubmitBid(types.Bid) error {
+	return nil
+}
+
 func TestOwnMessageDropping(t *testing.T) {
 	var wasCalled *bool = new(bool)
 	fakeStore := FakeStoreBackend{OnStore: func(bid Bid, caller common.Address, key string, value []byte) (Bid, error) {
@@ -50,7 +62,7 @@ func TestOwnMessageDropping(t *testing.T) {
 	}}
 
 	fakeDaSigner := FakeDASigner{localAddresses: []common.Address{{0x42}}}
-	engine, err := NewConfidentialStoreEngine(&fakeStore, MockTransport{}, MockMempool{}, fakeDaSigner, MockChainSigner{})
+	engine, err := NewConfidentialStoreEngine(&fakeStore, MockTransport{}, fakeDaSigner, MockChainSigner{})
 	require.NoError(t, err)
 
 	dummyCreationTx := types.NewTx(&types.ConfidentialComputeRequest{

suave/core/types.go

@@ -36,17 +36,16 @@ var ErrBidAlreadyPresent = errors.New("bid already present")
 
 type ConfidentialStoreBackend interface {
 	node.Lifecycle
+
 	InitializeBid(bid Bid) error
 	FetchEngineBidById(bidId BidId) (Bid, error)
 	Store(bid Bid, caller common.Address, key string, value []byte) (Bid, error)
 	Retrieve(bid Bid, caller common.Address, key string) ([]byte, error)
-}
-
-type MempoolBackend interface {
-	node.Lifecycle
-	SubmitBid(types.Bid) error
 	FetchBidById(BidId) (types.Bid, error)
 	FetchBidsByProtocolAndBlock(blockNumber uint64, namespace string) []types.Bid
+
+	// TODO: remove this
+	SubmitBid(types.Bid) error
 }
 
 type ConfidentialEthBackend interface {