Add partial mempool loading of requests

Author: lmoe

clients/iscmove/iscmoveclient/client_request.go

@@ -1,11 +1,14 @@
 package iscmoveclient
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"sort"
 	"time"
 
 	"github.com/samber/lo"
+	"golang.org/x/exp/maps"
 
 	"github.com/iotaledger/wasp/clients/iota-go/iotaclient"
 	"github.com/iotaledger/wasp/clients/iota-go/iotago"
@@ -212,10 +215,17 @@ func (c *Client) parseRequestAndFetchAssetsBag(obj *iotajsonrpc.IotaObjectData)
 	}, nil
 }
 
-func (c *Client) GetRequestsWithCB(ctx context.Context, packageID iotago.PackageID, anchorAddress *iotago.ObjectID, cb func(error, *iscmove.RefWithObject[iscmove.Request])) error {
+func (c *Client) GetRequestsSorted(ctx context.Context, packageID iotago.PackageID, anchorAddress *iotago.ObjectID, maxAmountOfRequests int, cb func(error, *iscmove.RefWithObject[iscmove.Request])) {
 	var lastSeen *iotago.ObjectID
+
+	pulledRequests := map[iotago.ObjectID]*iotajsonrpc.IotaObjectData{}
+	sortedRequestIDs := make([]iotago.ObjectID, 0)
+
+	_ = (pulledRequests)
+	_ = sortedRequestIDs
+
 	for {
-		res, err := c.GetOwnedObjects(ctx, iotaclient.GetOwnedObjectsRequest{
+		objs, err := c.GetOwnedObjects(ctx, iotaclient.GetOwnedObjectsRequest{
 			Address: anchorAddress,
 			Query: &iotajsonrpc.IotaObjectResponseQuery{
 				Filter: &iotajsonrpc.IotaObjectDataFilter{
@@ -229,29 +239,44 @@ func (c *Client) GetRequestsWithCB(ctx context.Context, packageID iotago.Package
 			},
 			Cursor: lastSeen,
 		})
+
 		if ctx.Err() != nil {
 			cb(fmt.Errorf("failed to fetch requests: %w", err), nil)
 			continue
 		}
-		if len(res.Data) == 0 {
-			return nil
+
+		if objs == nil || len(objs.Data) == 0 {
+			break
 		}
 
-		lastSeen = res.NextCursor
-		for _, reqData := range res.Data {
-			if ctx.Err() != nil {
-				return ctx.Err()
-			}
+		lastSeen = objs.NextCursor
 
-			fmt.Printf("Polling request id:%s, digest: %s, now:%s\n", reqData.Data.ObjectID, reqData.Data.Digest, time.Now().String())
-			req, err := c.parseRequestAndFetchAssetsBag(reqData.Data)
-			if err != nil {
-				cb(fmt.Errorf("failed to decode requests: %w", err), nil)
-			} else {
-				cb(nil, req)
-			}
+		for _, req := range objs.Data {
+			pulledRequests[*req.Data.ObjectID] = req.Data
 		}
 	}
+
+	// Sort object IDs
+	objectKeys := maps.Keys(pulledRequests)
+	sort.Slice(objectKeys, func(i, j int) bool {
+		return bytes.Compare(objectKeys[i][:], objectKeys[j][:]) < 0
+	})
+
+	if len(objectKeys) >= maxAmountOfRequests {
+		sortedRequestIDs = objectKeys[:maxAmountOfRequests]
+	} else {
+		sortedRequestIDs = objectKeys
+	}
+
+	// We only pass the maxRequests amount of requests into the result, to not overload the mempool.
+	// Periodically, this function will be called by the Chain Manager to pick up the next amount of transactions.
+	// This ensures that we don't suddenly load 20k of Requests into the mempool which fills up at a lower limit.
+	// It improves the startup time of the node and makes sure that the Consensus gets the same IDs across all nodes in a sorted manner.
+
+	for _, reqID := range sortedRequestIDs {
+		ref, err := c.parseRequestAndFetchAssetsBag(pulledRequests[reqID])
+		cb(err, ref)
+	}
 }
 
 func (c *Client) GetRequests(

clients/iscmove/iscmoveclient/feed.go

@@ -51,15 +51,15 @@ func (f *ChainFeed) GetCurrentAnchor(ctx context.Context) (*iscmove.AnchorWithRe
 
 // FetchCurrentState fetches the current Anchor and all Requests owned by the
 // anchor address.
-func (f *ChainFeed) FetchCurrentState(ctx context.Context, requestCb func(error, *iscmove.RefWithObject[iscmove.Request])) (*iscmove.AnchorWithRef, error) {
+func (f *ChainFeed) FetchCurrentState(ctx context.Context, maxAmountOfRequests int, requestCb func(error, *iscmove.RefWithObject[iscmove.Request])) (*iscmove.AnchorWithRef, error) {
 	anchor, err := f.GetCurrentAnchor(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	go f.wsClient.GetRequestsWithCB(ctx, f.iscPackageID, &f.anchorAddress, requestCb)
+	f.wsClient.GetRequestsSorted(ctx, f.iscPackageID, &f.anchorAddress, maxAmountOfRequests, requestCb)
 
-	return anchor, nil
+	return anchor, err
 }
 
 // SubscribeToUpdates starts fetching updated versions of the Anchor and newly received requests in background.

components/nodeconn/component.go

@@ -10,7 +10,9 @@ import (
 
 	"github.com/iotaledger/hive.go/app"
 	"github.com/iotaledger/hive.go/app/shutdown"
+
 	"github.com/iotaledger/wasp/clients/iota-go/iotago"
+	"github.com/iotaledger/wasp/components/chains"
 	"github.com/iotaledger/wasp/packages/chain"
 	"github.com/iotaledger/wasp/packages/daemon"
 	"github.com/iotaledger/wasp/packages/nodeconn"
@@ -51,6 +53,7 @@ func provide(c *dig.Container) error {
 		nodeConnection, err := nodeconn.New(
 			Component.Daemon().ContextStopped(),
 			*address,
+			chains.ParamsChains.MempoolMaxOnledgerInPool,
 			ParamsWS.WebsocketURL,
 			Component.Logger().Named("nc"),
 			deps.ShutdownHandler,

packages/nodeconn/chain.go

@@ -201,7 +201,10 @@ func (ncc *ncChain) syncChainState(ctx context.Context) error {
 		return err
 	}
 
-	_, err = ncc.feed.FetchCurrentState(ctx, func(err error, req *iscmove.RefWithObject[iscmove.Request]) {
+	anchor := isc.NewStateAnchor(moveAnchor, ncc.feed.GetISCPackageID())
+	ncc.anchorHandler(&anchor)
+
+	go ncc.feed.FetchCurrentState(ctx, ncc.nodeConn.maxNumberOfRequests, func(err error, req *iscmove.RefWithObject[iscmove.Request]) {
 		if err != nil {
 			return
 		}
@@ -210,14 +213,10 @@ func (ncc *ncChain) syncChainState(ctx context.Context) error {
 		if err != nil {
 			return
 		}
+
+		ncc.LogInfof("Sending %s to request handler", req.ObjectID)
 		ncc.requestHandler(onledgerReq)
 	})
-	if err != nil {
-		return err
-	}
-
-	anchor := isc.NewStateAnchor(moveAnchor, ncc.feed.GetISCPackageID())
-	ncc.anchorHandler(&anchor)
 
 	ncc.LogInfof("Synchronizing chain state for %s... done", ncc.chainID)
 	return nil

packages/nodeconn/nodeconn.go

@@ -20,6 +20,7 @@ import (
 	"github.com/iotaledger/hive.go/app/shutdown"
 	"github.com/iotaledger/hive.go/ds/shrinkingmap"
 	"github.com/iotaledger/hive.go/logger"
+
 	"github.com/iotaledger/wasp/clients/iota-go/iotago"
 	"github.com/iotaledger/wasp/clients/iota-go/iotasigner"
 	"github.com/iotaledger/wasp/clients/iscmove/iscmoveclient"
@@ -60,9 +61,10 @@ type nodeConnection struct {
 	iscPackageID iotago.PackageID
 	wsClient     *iscmoveclient.Client
 
-	synced     sync.WaitGroup
-	chainsLock sync.RWMutex
-	chainsMap  *shrinkingmap.ShrinkingMap[isc.ChainID, *ncChain]
+	maxNumberOfRequests int
+	synced              sync.WaitGroup
+	chainsLock          sync.RWMutex
+	chainsMap           *shrinkingmap.ShrinkingMap[isc.ChainID, *ncChain]
 
 	shutdownHandler *shutdown.ShutdownHandler
 }
@@ -72,6 +74,7 @@ var _ chain.NodeConnection = &nodeConnection{}
 func New(
 	ctx context.Context,
 	iscPackageID iotago.PackageID,
+	maxNumberOfRequests int,
 	wsURL string,
 	log *logger.Logger,
 	shutdownHandler *shutdown.ShutdownHandler,
@@ -81,9 +84,10 @@ func New(
 		return nil, err
 	}
 	return &nodeConnection{
-		WrappedLogger: logger.NewWrappedLogger(log),
-		iscPackageID:  iscPackageID,
-		wsClient:      wsClient,
+		WrappedLogger:       logger.NewWrappedLogger(log),
+		iscPackageID:        iscPackageID,
+		wsClient:            wsClient,
+		maxNumberOfRequests: maxNumberOfRequests,
 		chainsMap: shrinkingmap.New[isc.ChainID, *ncChain](
 			shrinkingmap.WithShrinkingThresholdRatio(chainsCleanupThresholdRatio),
 			shrinkingmap.WithShrinkingThresholdCount(chainsCleanupThresholdCount),