feat(x/async): EndBlocker schedules pending Future

Implement an EndBlocker that takes all the Futures that do not have a
result yet, and schedule them to the prophet's process for asynchronous
execution.

Author: Pitasi

warden/x/async/keeper/abci.go

@@ -5,13 +5,32 @@ import (
 
 	cometabci "github.com/cometbft/cometbft/abci/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/warden-protocol/wardenprotocol/prophet"
 )
 
 func (k Keeper) BeginBlocker(ctx context.Context) error {
 	return nil
 }
 
+// EndBlocker schedules some pending futures to the Prophet's process.
+//
+// Note: if a future remains pending for more blocks, it could be re-added to
+// Prophet even if it's already in the Prophet's queue or it's being processed.
+// This is not a problem as Prophet filters out incoming duplicate futures.
 func (k Keeper) EndBlocker(ctx context.Context) error {
+	futures, err := k.futures.PendingFutures(ctx, 10)
+	if err != nil {
+		return err
+	}
+
+	for _, f := range futures {
+		k.p.AddFuture(prophet.Future{
+			ID:      f.Id,
+			Handler: f.Handler,
+			Input:   f.Input,
+		})
+	}
+
 	return nil
 }
 

warden/x/async/keeper/futures.go

@@ -16,6 +16,7 @@ type FutureKeeper struct {
 	futures         repo.SeqCollection[types.Future]
 	futureByCreator collections.KeySet[collections.Pair[sdk.AccAddress, uint64]]
 	results         collections.Map[uint64, types.FutureResult]
+	pendingFutures  collections.KeySet[uint64]
 }
 
 func NewFutureKeeper(sb *collections.SchemaBuilder, cdc codec.Codec) *FutureKeeper {
@@ -27,10 +28,13 @@ func NewFutureKeeper(sb *collections.SchemaBuilder, cdc codec.Codec) *FutureKeep
 
 	results := collections.NewMap(sb, ResultsPrefix, "future_results", collections.Uint64Key, codec.CollValue[types.FutureResult](cdc))
 
+	pendingFutures := collections.NewKeySet(sb, PendingFuturesPrefix, "pending_futures", collections.Uint64Key)
+
 	return &FutureKeeper{
 		futures:         futures,
 		futureByCreator: futureByCreator,
 		results:         results,
+		pendingFutures:  pendingFutures,
 	}
 }
 
@@ -49,6 +53,10 @@ func (k *FutureKeeper) Append(ctx context.Context, t *types.Future) (uint64, err
 		return 0, err
 	}
 
+	if err := k.pendingFutures.Set(ctx, id); err != nil {
+		return 0, err
+	}
+
 	return id, nil
 }
 
@@ -61,6 +69,9 @@ func (k *FutureKeeper) Set(ctx context.Context, f types.Future) error {
 }
 
 func (k *FutureKeeper) SetResult(ctx context.Context, result types.FutureResult) error {
+	if err := k.pendingFutures.Remove(ctx, result.Id); err != nil {
+		return err
+	}
 	return k.results.Set(ctx, result.Id, result)
 }
 
@@ -75,3 +86,31 @@ func (k *FutureKeeper) HasResult(ctx context.Context, id uint64) (bool, error) {
 func (k *FutureKeeper) Futures() repo.SeqCollection[types.Future] {
 	return k.futures
 }
+
+func (k *FutureKeeper) PendingFutures(ctx context.Context, limit int) ([]types.Future, error) {
+	it, err := k.pendingFutures.IterateRaw(ctx, nil, nil, collections.OrderAscending)
+	if err != nil {
+		return nil, err
+	}
+	defer it.Close()
+
+	futures := make([]types.Future, 0, limit)
+	for ; it.Valid(); it.Next() {
+		id, err := it.Key()
+		if err != nil {
+			return nil, err
+		}
+
+		fut, err := k.futures.Get(ctx, id)
+		if err != nil {
+			return nil, err
+		}
+
+		futures = append(futures, fut)
+		if len(futures) == limit {
+			break
+		}
+	}
+
+	return futures, nil
+}

warden/x/async/keeper/keeper.go

@@ -37,6 +37,7 @@ var (
 	FutureByAddressPrefix = collections.NewPrefix(2)
 	ResultsPrefix         = collections.NewPrefix(3)
 	VotesPrefix           = collections.NewPrefix(4)
+	PendingFuturesPrefix  = collections.NewPrefix(5)
 )
 
 func NewKeeper(