Merge pull request #3518 from iotaledger/fix-mp112233

Do not forward rejected requests.

Author: lmoe

packages/chain/mempool/mempool.go

@@ -52,6 +52,7 @@ import (
 	"github.com/samber/lo"
 
 	"github.com/iotaledger/hive.go/logger"
+
 	consGR "github.com/iotaledger/wasp/packages/chain/cons/cons_gr"
 	"github.com/iotaledger/wasp/packages/chain/mempool/distsync"
 	"github.com/iotaledger/wasp/packages/cryptolib"
@@ -505,8 +506,9 @@ func (mpi *mempoolImpl) distSyncRequestReceivedCB(request isc.Request) bool {
 		return false
 	}
 	if err := mpi.shouldAddOffledgerRequest(offLedgerReq); err == nil {
-		mpi.addOffledger(offLedgerReq)
-		return true
+		mpi.log.Warn("shouldAddOffledgerRequest: true, trying to add to offledger %T: %+v", request, request)
+
+		return mpi.addOffledger(offLedgerReq)
 	}
 	return false
 }
@@ -558,10 +560,13 @@ func (mpi *mempoolImpl) shouldAddOffledgerRequest(req isc.OffLedgerRequest) erro
 	return nil
 }
 
-func (mpi *mempoolImpl) addOffledger(request isc.OffLedgerRequest) {
-	mpi.offLedgerPool.Add(request)
+func (mpi *mempoolImpl) addOffledger(request isc.OffLedgerRequest) bool {
+	if !mpi.offLedgerPool.Add(request) {
+		return false
+	}
 	mpi.metrics.IncRequestsReceived(request)
 	mpi.log.Debugf("accepted by the mempool, requestID: %s", request.ID().String())
+	return true
 }
 
 func (mpi *mempoolImpl) handleServerNodesUpdated(recv *reqServerNodesUpdated) {
@@ -806,8 +811,9 @@ func (mpi *mempoolImpl) handleReceiveOnLedgerRequest(request isc.OnLedgerRequest
 
 func (mpi *mempoolImpl) handleReceiveOffLedgerRequest(request isc.OffLedgerRequest) {
 	mpi.log.Debugf("Received request %v from outside.", request.ID())
-	mpi.addOffledger(request)
-	mpi.sendMessages(mpi.distSync.Input(distsync.NewInputPublishRequest(request)))
+	if mpi.addOffledger(request) {
+		mpi.sendMessages(mpi.distSync.Input(distsync.NewInputPublishRequest(request)))
+	}
 }
 
 func (mpi *mempoolImpl) handleTangleTimeUpdated(tangleTime time.Time) {

packages/chain/mempool/offledger_pool.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/iotaledger/hive.go/ds/shrinkingmap"
 	"github.com/iotaledger/hive.go/logger"
+
 	consGR "github.com/iotaledger/wasp/packages/chain/cons/cons_gr"
 	"github.com/iotaledger/wasp/packages/isc"
 	"github.com/iotaledger/wasp/packages/kv/codec"
@@ -67,16 +68,16 @@ func (p *OffLedgerPool) Get(reqRef *isc.RequestRef) isc.OffLedgerRequest {
 	return entry.req
 }
 
-func (p *OffLedgerPool) Add(request isc.OffLedgerRequest) {
+func (p *OffLedgerPool) Add(request isc.OffLedgerRequest) bool {
 	ref := isc.RequestRefFromRequest(request)
 	entry := &OrderedPoolEntry{req: request, ts: time.Now()}
 	account := request.SenderAccount().String()
 
 	//
 	// add the request to the "request ref" Lookup Table
 	if !p.refLUT.Set(ref.AsKey(), entry) {
-		p.log.Debugf("NOT ADDED, already exists. reqID: %v as key=%v, senderAccount: ", request.ID(), ref, account)
-		return // not added already exists
+		p.log.Debugf("OffLedger Request NOT ADDED, already exists. reqID: %v as key=%v, senderAccount: %v", request.ID(), ref, account)
+		return true // not added already exists
 	}
 
 	//
@@ -134,14 +135,16 @@ func (p *OffLedgerPool) Add(request isc.OffLedgerRequest) {
 	deleted := p.LimitPoolSize()
 	if lo.Contains(deleted, entry) {
 		// this exact request was deleted from the pool, do not update metrics, or mark available
-		return
+		p.log.Debugf("OffLedger Request NOT ADDED, was removed already. reqID: %v as key=%v, senderAccount: %v", request.ID(), ref, account)
+		return false
 	}
 
 	//
 	// update metrics and signal that the request is available
 	p.log.Debugf("ADD %v as key=%v, senderAccount: %s", request.ID(), ref, account)
 	p.sizeMetric(p.refLUT.Size())
 	p.waitReq.MarkAvailable(request)
+	return true
 }
 
 // LimitPoolSize drops the txs with the lowest price if the total number of requests is too big