[CP-341] Async CheckTx and Unified Mempool

abci/client/unsync_local_client.go

@@ -44,23 +44,26 @@ func (app *unsyncLocalClient) SetResponseCallback(cb Callback) {
 }
 
 func (app *unsyncLocalClient) CheckTxAsync(ctx context.Context, req *types.CheckTxRequest) (*ReqRes, error) {
-	res, err := app.Application.CheckTx(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-	return app.callback(
-		types.ToCheckTxRequest(req),
-		types.ToCheckTxResponse(res),
-	), nil
-}
+	reqres := NewReqRes(types.ToCheckTxRequest(req))
 
-func (app *unsyncLocalClient) callback(req *types.Request, res *types.Response) *ReqRes {
-	if app.Callback != nil {
-		app.Callback(req, res)
-	}
-	rr := newLocalReqRes(req, res)
-	rr.callbackInvoked = true
-	return rr
+	go func() {
+		res, err := app.Application.CheckTx(ctx, req)
+		if err != nil {
+			reqres.Response = types.ToExceptionResponse("") // optimistic recheck failed
+		} else {
+			reqres.Response = types.ToCheckTxResponse(res)
+		}
+
+		if app.Callback != nil {
+			app.Callback(reqres.Request, reqres.Response)
+		}
+
+		reqres.Done()
+
+		reqres.InvokeCallback()
+	}()
+
+	return reqres, nil
 }
 
 // -------------------------------------------------------

config/config.go

@@ -49,6 +49,7 @@ const (
 	v2 = "v2"
 
 	MempoolTypeFlood = "flood"
+	MempoolTypeProxy = "proxy"
 	MempoolTypeNop   = "nop"
 )
 

mempool/mempool.go

@@ -181,3 +181,21 @@ type Iterator interface {
 	// WaitNextCh returns a channel on which to wait for the next available entry.
 	WaitNextCh() <-chan Entry
 }
+
+// MempoolTx defines the interface for a transaction in the mempool
+// It provides methods to access transaction properties and sender information
+type MempoolTx interface {
+	Height() int64
+	GasWanted() int64
+	Tx() types.Tx
+	IsSender(peerID p2p.ID) bool
+	AddSender(peerID p2p.ID) bool
+	Senders() []p2p.ID
+}
+
+// TxBroadcastStream defines the interface for streaming transactions to broadcast.
+// It provides a channel that will receive transactions to be broadcasted to peers.
+type TxBroadcastStream interface {
+	// GetTxChannel returns a channel that will receive transactions to broadcast.
+	GetTxChannel() <-chan MempoolTx
+}

mempool/mempoolTx.go

@@ -10,6 +10,9 @@ import (
 )
 
 // mempoolTx is an entry in the mempool.
+var _ MempoolTx = (*mempoolTx)(nil)
+
+// mempoolTx is an entry in the mempool
 type mempoolTx struct {
 	height    int64    // height that this tx had been validated in
 	gasWanted int64    // amount of gas this tx states it will require
@@ -19,10 +22,17 @@ type mempoolTx struct {
 	timestamp time.Time // time when entry was created
 
 	// ids of peers who've sent us this tx (as a map for quick lookups).
-	// senders: PeerID -> struct{}
+	// senders: PeerID -> bool
 	senders sync.Map
 }
 
+// NewMempoolTx creates a new mempoolTx using the builder pattern
+func NewMempoolTx(tx types.Tx) MempoolTx {
+	return NewMempoolTxBuilder().
+		WithTx(tx).
+		Build()
+}
+
 func (memTx *mempoolTx) Tx() types.Tx {
 	return memTx.tx
 }
@@ -32,14 +42,18 @@ func (memTx *mempoolTx) Height() int64 {
 }
 
 func (memTx *mempoolTx) GasWanted() int64 {
-	return memTx.gasWanted
+	return atomic.LoadInt64(&memTx.gasWanted)
 }
 
 func (memTx *mempoolTx) IsSender(peerID p2p.ID) bool {
 	_, ok := memTx.senders.Load(peerID)
 	return ok
 }
 
+func (memTx *mempoolTx) AddSender(peerID p2p.ID) bool {
+	return memTx.addSender(peerID)
+}
+
 // Add the peer ID to the list of senders. Return true iff it exists already in the list.
 func (memTx *mempoolTx) addSender(peerID p2p.ID) bool {
 	if len(peerID) == 0 {
@@ -59,3 +73,54 @@ func (memTx *mempoolTx) Senders() []p2p.ID {
 	})
 	return senders
 }
+
+// MempoolTxBuilder is a builder for creating mempoolTx instances
+type MempoolTxBuilder struct {
+	height    int64
+	gasWanted int64
+	tx        types.Tx
+	senders   []p2p.ID
+}
+
+// NewMempoolTxBuilder creates a new builder for mempoolTx
+func NewMempoolTxBuilder() *MempoolTxBuilder {
+	return &MempoolTxBuilder{
+		senders: make([]p2p.ID, 0),
+	}
+}
+
+// WithHeight sets the height for the mempoolTx
+func (b *MempoolTxBuilder) WithHeight(height int64) *MempoolTxBuilder {
+	b.height = height
+	return b
+}
+
+// WithGasWanted sets the gas wanted for the mempoolTx
+func (b *MempoolTxBuilder) WithGasWanted(gasWanted int64) *MempoolTxBuilder {
+	b.gasWanted = gasWanted
+	return b
+}
+
+// WithTx sets the transaction for the mempoolTx
+func (b *MempoolTxBuilder) WithTx(tx types.Tx) *MempoolTxBuilder {
+	b.tx = tx
+	return b
+}
+
+func (b *MempoolTxBuilder) WithSender(sender p2p.ID) *MempoolTxBuilder {
+	b.senders = append(b.senders, sender)
+	return b
+}
+
+// Build creates the final mempoolTx instance
+func (b *MempoolTxBuilder) Build() MempoolTx {
+	memTx := &mempoolTx{
+		height:    b.height,
+		gasWanted: b.gasWanted,
+		tx:        b.tx,
+	}
+	for _, sender := range b.senders {
+		memTx.senders.Store(sender, struct{}{})
+	}
+	return memTx
+}