feat: added PendingOutbound queries in uexecutor module

app/upgrades.go

@@ -26,6 +26,7 @@ import (
 	tsscoreevmparamsfix "github.com/pushchain/push-chain-node/app/upgrades/tss-core-evm-params-fix"
 	tsscorefix "github.com/pushchain/push-chain-node/app/upgrades/tss-core-fix"
 	tssvotegasless "github.com/pushchain/push-chain-node/app/upgrades/tss-vote-gasless"
+	pendingoutboundsindex "github.com/pushchain/push-chain-node/app/upgrades/pending-outbounds-index"
 	universaltxv1 "github.com/pushchain/push-chain-node/app/upgrades/universal-tx-v1"
 )
 
@@ -51,6 +52,7 @@ var Upgrades = []upgrades.Upgrade{
 	chainmetavotegasless.NewUpgrade(),
 	ceagasandpayload.NewUpgrade(),
 	ceapayloadverificationfix.NewUpgrade(),
+	pendingoutboundsindex.NewUpgrade(),
 }
 
 // RegisterUpgradeHandlers registers the chain upgrade handlers

app/upgrades/pending-outbounds-index/upgrade.go

@@ -0,0 +1,96 @@
+package pendingoutboundsindex
+
+import (
+	"context"
+
+	storetypes "cosmossdk.io/store/types"
+	upgradetypes "cosmossdk.io/x/upgrade/types"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/cosmos/cosmos-sdk/types/module"
+
+	"github.com/pushchain/push-chain-node/app/upgrades"
+	"github.com/pushchain/push-chain-node/x/uexecutor/types"
+)
+
+const UpgradeName = "pending-outbounds-index"
+
+// NewUpgrade constructs the upgrade definition
+func NewUpgrade() upgrades.Upgrade {
+	return upgrades.Upgrade{
+		UpgradeName:          UpgradeName,
+		CreateUpgradeHandler: CreateUpgradeHandler,
+		StoreUpgrades: storetypes.StoreUpgrades{
+			Added:   []string{},
+			Deleted: []string{},
+		},
+	}
+}
+
+func CreateUpgradeHandler(
+	mm upgrades.ModuleManager,
+	configurator module.Configurator,
+	ak *upgrades.AppKeepers,
+) upgradetypes.UpgradeHandler {
+	return func(ctx context.Context, plan upgradetypes.Plan, fromVM module.VersionMap) (module.VersionMap, error) {
+		sdkCtx := sdk.UnwrapSDKContext(ctx)
+		logger := sdkCtx.Logger().With("upgrade", UpgradeName)
+		logger.Info("Starting upgrade handler: backfilling PendingOutbounds index")
+
+		keeper := ak.UExecutorKeeper
+		count := 0
+
+		// Iterate all UniversalTx entries and backfill pending outbounds
+		iter, err := keeper.UniversalTx.Iterate(ctx, nil)
+		if err != nil {
+			logger.Error("Failed to create UniversalTx iterator", "error", err)
+			return nil, err
+		}
+		defer iter.Close()
+
+		for ; iter.Valid(); iter.Next() {
+			kv, err := iter.KeyValue()
+			if err != nil {
+				logger.Error("Failed to read UniversalTx entry", "error", err)
+				return nil, err
+			}
+
+			utxId := kv.Key
+			utx := kv.Value
+
+			for _, ob := range utx.OutboundTx {
+				if ob == nil {
+					continue
+				}
+				if ob.OutboundStatus == types.Status_PENDING {
+					entry := types.PendingOutboundEntry{
+						OutboundId:    ob.Id,
+						UniversalTxId: utxId,
+						CreatedAt:     0, // unknown historical height
+					}
+					if err := keeper.PendingOutbounds.Set(ctx, ob.Id, entry); err != nil {
+						logger.Error("Failed to set pending outbound", "outbound_id", ob.Id, "error", err)
+						return nil, err
+					}
+					count++
+				}
+			}
+
+			// Log progress every 1000 UTXs
+			if count > 0 && count%1000 == 0 {
+				logger.Info("Backfill progress", "pending_outbounds_indexed", count)
+			}
+		}
+
+		logger.Info("PendingOutbounds backfill complete", "total_indexed", count)
+
+		versionMap, err := mm.RunMigrations(ctx, configurator, fromVM)
+		if err != nil {
+			logger.Error("RunMigrations failed", "error", err)
+			return nil, err
+		}
+
+		logger.Info("Upgrade complete", "upgrade", UpgradeName)
+		return versionMap, nil
+	}
+}

proto/uexecutor/v1/query.proto

@@ -50,6 +50,16 @@ service Query {
   rpc AllChainMetas(QueryAllChainMetasRequest) returns (QueryAllChainMetasResponse) {
     option (google.api.http).get = "/uexecutor/v1/chain_metas";
   }
+
+  // Get a single pending outbound by ID
+  rpc GetPendingOutbound(QueryGetPendingOutboundRequest) returns (QueryGetPendingOutboundResponse) {
+    option (google.api.http).get = "/uexecutor/v1/pending_outbound/{outbound_id}";
+  }
+
+  // Get all pending outbounds (paginated)
+  rpc AllPendingOutbounds(QueryAllPendingOutboundsRequest) returns (QueryAllPendingOutboundsResponse) {
+    option (google.api.http).get = "/uexecutor/v1/pending_outbounds";
+  }
 }
 
 // ==========================
@@ -130,3 +140,29 @@ message QueryAllUniversalTxResponse {
   repeated UniversalTx universal_txs = 1;
   cosmos.base.query.v1beta1.PageResponse pagination = 2;
 }
+
+// Pending outbound index entry
+message PendingOutboundEntry {
+  string outbound_id     = 1;
+  string universal_tx_id = 2;
+  int64  created_at      = 3;
+}
+
+message QueryGetPendingOutboundRequest {
+  string outbound_id = 1;
+}
+
+message QueryGetPendingOutboundResponse {
+  PendingOutboundEntry entry    = 1;
+  OutboundTx           outbound = 2;
+}
+
+message QueryAllPendingOutboundsRequest {
+  cosmos.base.query.v1beta1.PageRequest pagination = 1;
+}
+
+message QueryAllPendingOutboundsResponse {
+  repeated PendingOutboundEntry entries   = 1;
+  repeated OutboundTx           outbounds = 2;
+  cosmos.base.query.v1beta1.PageResponse pagination = 3;
+}

x/uexecutor/autocli.go

@@ -43,6 +43,19 @@ func (am AppModule) AutoCLIOptions() *autocliv1.ModuleOptions {
 					Use:       "all-chain-metas",
 					Short:     "Query chain metadata for all chains",
 				},
+				{
+					RpcMethod: "GetPendingOutbound",
+					Use:       "pending-outbound [outbound-id]",
+					Short:     "Query a single pending outbound by ID",
+					PositionalArgs: []*autocliv1.PositionalArgDescriptor{
+						{ProtoField: "outbound_id"},
+					},
+				},
+				{
+					RpcMethod: "AllPendingOutbounds",
+					Use:       "all-pending-outbounds",
+					Short:     "Query all pending outbounds (paginated)",
+				},
 			},
 			SubCommands: map[string]*autocliv1.ServiceCommandDescriptor{
 				"v2": {

x/uexecutor/keeper/create_outbound.go

@@ -330,6 +330,15 @@ func (k Keeper) attachOutboundsToUtx(
 
 			utx.OutboundTx = append(utx.OutboundTx, outbound)
 
+			// Write to pending outbounds index (inside UpdateUniversalTx closure for atomicity)
+			if err := k.PendingOutbounds.Set(ctx, outbound.Id, types.PendingOutboundEntry{
+				OutboundId:    outbound.Id,
+				UniversalTxId: utxId,
+				CreatedAt:     ctx.BlockHeight(),
+			}); err != nil {
+				return fmt.Errorf("failed to set pending outbound index for %s: %w", outbound.Id, err)
+			}
+
 			var pcTxHash string
 			var logIndex string
 

x/uexecutor/keeper/keeper.go

@@ -52,6 +52,10 @@ type Keeper struct {
 
 	// ChainMetas collection stores aggregated chain metadata (gas price + block height) for each chain
 	ChainMetas collections.Map[string, types.ChainMeta]
+
+	// PendingOutbounds is a secondary index of outbounds with PENDING status.
+	// Key: outbound ID -> Value: PendingOutboundEntry
+	PendingOutbounds collections.Map[string, types.PendingOutboundEntry]
 }
 
 // NewKeeper creates a new Keeper instance
@@ -129,6 +133,14 @@ func NewKeeper(
 			collections.StringKey,
 			codec.CollValue[types.ChainMeta](cdc),
 		),
+
+		PendingOutbounds: collections.NewMap(
+			sb,
+			types.PendingOutboundsKey,
+			types.PendingOutboundsName,
+			collections.StringKey,
+			codec.CollValue[types.PendingOutboundEntry](cdc),
+		),
 	}
 
 	return k

x/uexecutor/keeper/msg_vote_outbound.go

@@ -80,7 +80,12 @@ func (k Keeper) VoteOutbound(
 		return err
 	}
 
-	// Step 6: Finalize outbound (refund if failed).
+	// Remove from pending outbounds index now that status is OBSERVED
+	if err := k.PendingOutbounds.Remove(ctx, outboundId); err != nil {
+		return fmt.Errorf("failed to remove pending outbound index for %s: %w", outboundId, err)
+	}
+
+	// Step 6: Finalize outbound (refund if failed) - Don't return error
 	// If the revert re-mint fails, handleFailedOutbound marks it ABORTED internally.
 	_ = k.FinalizeOutbound(ctx, utxId, outbound)