New mempool impl

Signed-off-by: Alessandro Sorniotti <aso@zurich.ibm.com>

Author: ale-linux

gateway/core/api.go

@@ -36,6 +36,32 @@ type Submitter interface {
 	Close() error
 }
 
+// TxQueueInterface defines the interface that transaction queue implementations must satisfy.
+// This allows switching between different queue implementations (e.g., TxQueue and TxQueueV2).
+type TxQueueInterface interface {
+	// Enqueue adds a transaction to the queue
+	Enqueue(tx *types.Transaction)
+
+	// Dequeue removes and returns a transaction from the queue
+	// Returns (transaction, true) if successful, or (nil, false) if queue is closed
+	Dequeue() (*types.Transaction, bool)
+
+	// IsPending checks if a transaction is currently in the queue or being processed
+	IsPending(txHash common.Hash) *types.Transaction
+
+	// Complete marks a transaction as completed
+	Complete(hash common.Hash)
+
+	// Close signals shutdown of the queue
+	Close()
+
+	// Handle processes block notifications from the synchronizer
+	Handle(ctx context.Context, block *domain.Block) error
+
+	// Stats returns statistics about processed transactions (total, invalid)
+	Stats() (total int, invalid int)
+}
+
 var logger = flogging.MustGetLogger("gateway.core")
 
 // Gateway is the component that bridges Fabric-x and the EVM. Its API is the
@@ -49,7 +75,7 @@ type Gateway struct {
 	chainID     *big.Int
 	ChainConfig *params.ChainConfig
 	Signer      types.Signer
-	TxQueue     *TxQueue
+	TxQueue     TxQueueInterface
 	workerCount int
 	wg          sync.WaitGroup
 	stopOnce    sync.Once
@@ -84,7 +110,7 @@ func New(ec *EndorsementClient, submitter Submitter, store Store, chainID int64,
 		chainID:     cid,
 		ChainConfig: cmn.BuildChainConfig(chainID),
 		Signer:      types.LatestSignerForChainID(cid),
-		TxQueue:     NewTxQueue(),
+		TxQueue:     NewTxQueueV2(),
 		workerCount: workerCount,
 	}, nil
 }
@@ -346,7 +372,10 @@ func (g *Gateway) Stop() error {
 		err = g.submitter.Close()
 	})
 
-	fmt.Println("gw stats:", g.TxQueue.total, g.TxQueue.invalid, float64(g.TxQueue.invalid)/float64(g.TxQueue.total))
+	total, invalid := g.TxQueue.Stats()
+	if total > 0 {
+		fmt.Println("gw stats:", total, invalid, float64(invalid)/float64(total))
+	}
 
 	return err
 }

gateway/core/txqueue.go

@@ -222,55 +222,12 @@ func (q *TxQueue) Complete(hash common.Hash) {
 	}
 }
 
-func participantsForTx(tx *types.Transaction) []common.Address {
-	participants := make([]common.Address, 0, 2)
-
-	if sender, ok := senderForTx(tx); ok {
-		participants = append(participants, sender)
-	}
-
-	if recipient, ok := recipientForTx(tx); ok && !containsAddress(participants, recipient) {
-		participants = append(participants, recipient)
-	}
-
-	return participants
-}
-
-func senderForTx(tx *types.Transaction) (common.Address, bool) {
-	if !tx.Protected() && tx.Type() == types.LegacyTxType {
-		return common.Address{}, false
-	}
-
-	signer := types.LatestSignerForChainID(tx.ChainId())
-	sender, err := types.Sender(signer, tx)
-	if err != nil {
-		return common.Address{}, false
-	}
-
-	return sender, true
-}
-
-func recipientForTx(tx *types.Transaction) (common.Address, bool) {
-	if tx.To() == nil {
-		return common.Address{}, false
-	}
-
-	data := tx.Data()
-	if len(data) < 4+32+32 {
-		return common.Address{}, false
-	}
-
-	recipientOffset := 4 + 12
-	return common.BytesToAddress(data[recipientOffset : recipientOffset+20]), true
-}
-
-func containsAddress(addresses []common.Address, target common.Address) bool {
-	for _, address := range addresses {
-		if address == target {
-			return true
-		}
-	}
-	return false
+// Stats returns statistics about processed transactions.
+// Returns (total transactions processed, invalid transactions).
+func (q *TxQueue) Stats() (int, int) {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+	return q.total, q.invalid
 }
 
 // Handle processes block notifications from the synchronizer and marks transactions as complete.

gateway/core/txqueue_helpers.go

@@ -0,0 +1,74 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: LGPL-3.0-or-later
+*/
+
+package core
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+// participantsForTx extracts the sender and recipient addresses from a transaction.
+// Returns a slice containing unique participant addresses (1 or 2 elements).
+func participantsForTx(tx *types.Transaction) []common.Address {
+	participants := make([]common.Address, 0, 2)
+
+	if sender, ok := senderForTx(tx); ok {
+		participants = append(participants, sender)
+	}
+
+	if recipient, ok := recipientForTx(tx); ok && !containsAddress(participants, recipient) {
+		participants = append(participants, recipient)
+	}
+
+	return participants
+}
+
+// senderForTx extracts the sender address from a transaction.
+// Returns (address, true) if successful, (zero address, false) otherwise.
+func senderForTx(tx *types.Transaction) (common.Address, bool) {
+	if !tx.Protected() && tx.Type() == types.LegacyTxType {
+		return common.Address{}, false
+	}
+
+	signer := types.LatestSignerForChainID(tx.ChainId())
+	sender, err := types.Sender(signer, tx)
+	if err != nil {
+		return common.Address{}, false
+	}
+
+	return sender, true
+}
+
+// recipientForTx extracts the recipient address from a transaction.
+// For ERC20 transfers, it extracts the recipient from the calldata.
+// Returns (address, true) if successful, (zero address, false) otherwise.
+func recipientForTx(tx *types.Transaction) (common.Address, bool) {
+	if tx.To() == nil {
+		return common.Address{}, false
+	}
+
+	data := tx.Data()
+	if len(data) < 4+32+32 {
+		return common.Address{}, false
+	}
+
+	// Extract recipient from ERC20 transfer calldata (offset 4 + 12 bytes)
+	recipientOffset := 4 + 12
+	return common.BytesToAddress(data[recipientOffset : recipientOffset+20]), true
+}
+
+// containsAddress checks if a slice of addresses contains the target address.
+func containsAddress(addresses []common.Address, target common.Address) bool {
+	for _, address := range addresses {
+		if address == target {
+			return true
+		}
+	}
+	return false
+}
+
+// Made with Bob