diff --git a/app/upgrades.go b/app/upgrades.go index 09271a529..b66928bdf 100755 --- a/app/upgrades.go +++ b/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 diff --git a/app/upgrades/pending-outbounds-index/upgrade.go b/app/upgrades/pending-outbounds-index/upgrade.go new file mode 100644 index 000000000..d37e5008f --- /dev/null +++ b/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 + } +} diff --git a/proto/uexecutor/v1/query.proto b/proto/uexecutor/v1/query.proto index b8e9eb25b..e5e9eef50 100755 --- a/proto/uexecutor/v1/query.proto +++ b/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; +} diff --git a/x/uexecutor/autocli.go b/x/uexecutor/autocli.go index 2eca20605..a6f3b9fff 100755 --- a/x/uexecutor/autocli.go +++ b/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": { diff --git a/x/uexecutor/keeper/create_outbound.go b/x/uexecutor/keeper/create_outbound.go index 1021edffb..5dd7d3904 100644 --- a/x/uexecutor/keeper/create_outbound.go +++ b/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 diff --git a/x/uexecutor/keeper/keeper.go b/x/uexecutor/keeper/keeper.go index 63cf12dbe..f5c27dc73 100755 --- a/x/uexecutor/keeper/keeper.go +++ b/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 diff --git a/x/uexecutor/keeper/msg_vote_outbound.go b/x/uexecutor/keeper/msg_vote_outbound.go index 956fa9f54..cd0019696 100644 --- a/x/uexecutor/keeper/msg_vote_outbound.go +++ b/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) diff --git a/x/uexecutor/keeper/pending_outbound_test.go b/x/uexecutor/keeper/pending_outbound_test.go new file mode 100644 index 000000000..b4ce8bbdc --- /dev/null +++ b/x/uexecutor/keeper/pending_outbound_test.go @@ -0,0 +1,237 @@ +package keeper_test + +import ( + "fmt" + "testing" + + "github.com/golang/mock/gomock" + "github.com/pushchain/push-chain-node/x/uexecutor/types" + "github.com/stretchr/testify/require" +) + +func setupPendingOutboundFixture(t *testing.T) *testFixture { + t.Helper() + f := SetupTest(t) + + // Setup EVM mock for InitGenesis + f.mockEVMKeeper.EXPECT().SetAccount(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + f.mockEVMKeeper.EXPECT().SetCode(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes() + f.mockEVMKeeper.EXPECT().SetState(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes() + + f.k.InitGenesis(f.ctx, &types.GenesisState{Params: types.DefaultParams()}) + return f +} + +func TestPendingOutbound_IndexOnCreate(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + // Manually set a PendingOutbound entry (simulating what attachOutboundsToUtx does) + entry := types.PendingOutboundEntry{ + OutboundId: "outbound-1", + UniversalTxId: "utx-1", + CreatedAt: 100, + } + err := f.k.PendingOutbounds.Set(f.ctx, "outbound-1", entry) + require.NoError(err) + + // Verify it exists + got, err := f.k.PendingOutbounds.Get(f.ctx, "outbound-1") + require.NoError(err) + require.Equal("outbound-1", got.OutboundId) + require.Equal("utx-1", got.UniversalTxId) + require.Equal(int64(100), got.CreatedAt) +} + +func TestPendingOutbound_RemoveOnVote(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + // Set entry + err := f.k.PendingOutbounds.Set(f.ctx, "outbound-1", types.PendingOutboundEntry{ + OutboundId: "outbound-1", + UniversalTxId: "utx-1", + CreatedAt: 100, + }) + require.NoError(err) + + // Verify exists + has, err := f.k.PendingOutbounds.Has(f.ctx, "outbound-1") + require.NoError(err) + require.True(has) + + // Remove (simulating what VoteOutbound does) + err = f.k.PendingOutbounds.Remove(f.ctx, "outbound-1") + require.NoError(err) + + // Verify removed + has, err = f.k.PendingOutbounds.Has(f.ctx, "outbound-1") + require.NoError(err) + require.False(has) +} + +func TestPendingOutbound_GetPendingOutbound(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + // Create a UTX with an outbound + utx := types.UniversalTx{ + Id: "utx-1", + OutboundTx: []*types.OutboundTx{ + { + Id: "outbound-1", + DestinationChain: "eip155:1", + Recipient: "0xrecipient", + Amount: "1000", + Sender: "0xsender", + OutboundStatus: types.Status_PENDING, + }, + }, + } + require.NoError(f.k.UniversalTx.Set(f.ctx, "utx-1", utx)) + + // Index the pending outbound + require.NoError(f.k.PendingOutbounds.Set(f.ctx, "outbound-1", types.PendingOutboundEntry{ + OutboundId: "outbound-1", + UniversalTxId: "utx-1", + CreatedAt: 50, + })) + + // Query via querier + resp, err := f.queryServer.GetPendingOutbound(f.ctx, &types.QueryGetPendingOutboundRequest{ + OutboundId: "outbound-1", + }) + require.NoError(err) + require.NotNil(resp.Entry) + require.NotNil(resp.Outbound) + require.Equal("outbound-1", resp.Entry.OutboundId) + require.Equal("utx-1", resp.Entry.UniversalTxId) + require.Equal("eip155:1", resp.Outbound.DestinationChain) + require.Equal("0xrecipient", resp.Outbound.Recipient) + require.Equal("1000", resp.Outbound.Amount) +} + +func TestPendingOutbound_GetPendingOutbound_NotFound(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + _, err := f.queryServer.GetPendingOutbound(f.ctx, &types.QueryGetPendingOutboundRequest{ + OutboundId: "nonexistent", + }) + require.Error(err) + require.Contains(err.Error(), "not found") +} + +func TestPendingOutbound_GetPendingOutbound_EmptyId(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + _, err := f.queryServer.GetPendingOutbound(f.ctx, &types.QueryGetPendingOutboundRequest{ + OutboundId: "", + }) + require.Error(err) + require.Contains(err.Error(), "outbound_id is required") +} + +func TestPendingOutbound_AllPendingOutbounds(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + // Create 3 UTXs with outbounds + for i := 0; i < 3; i++ { + utxId := fmt.Sprintf("utx-%d", i) + outboundId := fmt.Sprintf("outbound-%d", i) + + utx := types.UniversalTx{ + Id: utxId, + OutboundTx: []*types.OutboundTx{ + { + Id: outboundId, + DestinationChain: fmt.Sprintf("eip155:%d", i+1), + Recipient: fmt.Sprintf("0xrecipient%d", i), + Amount: fmt.Sprintf("%d000", i+1), + Sender: "0xsender", + OutboundStatus: types.Status_PENDING, + }, + }, + } + require.NoError(f.k.UniversalTx.Set(f.ctx, utxId, utx)) + require.NoError(f.k.PendingOutbounds.Set(f.ctx, outboundId, types.PendingOutboundEntry{ + OutboundId: outboundId, + UniversalTxId: utxId, + CreatedAt: int64(i + 1), + })) + } + + // Query all + resp, err := f.queryServer.AllPendingOutbounds(f.ctx, &types.QueryAllPendingOutboundsRequest{}) + require.NoError(err) + require.Len(resp.Entries, 3) + require.Len(resp.Outbounds, 3) + + // Verify outbounds have full data + for _, ob := range resp.Outbounds { + require.NotEmpty(ob.DestinationChain) + require.NotEmpty(ob.Recipient) + require.NotEmpty(ob.Amount) + } +} + +func TestPendingOutbound_AllPendingOutbounds_Empty(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + resp, err := f.queryServer.AllPendingOutbounds(f.ctx, &types.QueryAllPendingOutboundsRequest{}) + require.NoError(err) + require.Empty(resp.Entries) +} + +func TestPendingOutbound_MultipleOutboundsPerUTX(t *testing.T) { + f := setupPendingOutboundFixture(t) + require := require.New(t) + + // Create a UTX with 2 outbounds + utx := types.UniversalTx{ + Id: "utx-multi", + OutboundTx: []*types.OutboundTx{ + { + Id: "outbound-a", + DestinationChain: "eip155:1", + Recipient: "0xrecipientA", + Amount: "1000", + OutboundStatus: types.Status_PENDING, + }, + { + Id: "outbound-b", + DestinationChain: "eip155:137", + Recipient: "0xrecipientB", + Amount: "2000", + OutboundStatus: types.Status_PENDING, + }, + }, + } + require.NoError(f.k.UniversalTx.Set(f.ctx, "utx-multi", utx)) + + // Index both + require.NoError(f.k.PendingOutbounds.Set(f.ctx, "outbound-a", types.PendingOutboundEntry{ + OutboundId: "outbound-a", UniversalTxId: "utx-multi", CreatedAt: 1, + })) + require.NoError(f.k.PendingOutbounds.Set(f.ctx, "outbound-b", types.PendingOutboundEntry{ + OutboundId: "outbound-b", UniversalTxId: "utx-multi", CreatedAt: 2, + })) + + // Query individual + respA, err := f.queryServer.GetPendingOutbound(f.ctx, &types.QueryGetPendingOutboundRequest{OutboundId: "outbound-a"}) + require.NoError(err) + require.Equal("eip155:1", respA.Outbound.DestinationChain) + + respB, err := f.queryServer.GetPendingOutbound(f.ctx, &types.QueryGetPendingOutboundRequest{OutboundId: "outbound-b"}) + require.NoError(err) + require.Equal("eip155:137", respB.Outbound.DestinationChain) + + // Query all — should return both + resp, err := f.queryServer.AllPendingOutbounds(f.ctx, &types.QueryAllPendingOutboundsRequest{}) + require.NoError(err) + require.Len(resp.Entries, 2) + require.Len(resp.Outbounds, 2) +} diff --git a/x/uexecutor/keeper/query_server.go b/x/uexecutor/keeper/query_server.go index 95de3c7da..f43a20ebc 100755 --- a/x/uexecutor/keeper/query_server.go +++ b/x/uexecutor/keeper/query_server.go @@ -324,6 +324,85 @@ func (k Querier) AllChainMetas(goCtx context.Context, req *types.QueryAllChainMe }, nil } +// GetPendingOutbound implements types.QueryServer. +// Returns a single pending outbound entry by ID, along with the full outbound data from the parent UTX. +func (k Querier) GetPendingOutbound(goCtx context.Context, req *types.QueryGetPendingOutboundRequest) (*types.QueryGetPendingOutboundResponse, error) { + if req == nil || req.OutboundId == "" { + return nil, status.Error(codes.InvalidArgument, "outbound_id is required") + } + + ctx := sdk.UnwrapSDKContext(goCtx) + + entry, err := k.PendingOutbounds.Get(ctx, req.OutboundId) + if err != nil { + if errors.Is(err, collections.ErrNotFound) { + return nil, status.Errorf(codes.NotFound, "pending outbound not found: %s", req.OutboundId) + } + return nil, status.Error(codes.Internal, err.Error()) + } + + // Fetch the parent UTX to get the full outbound data + utx, err := k.UniversalTx.Get(ctx, entry.UniversalTxId) + if err != nil { + if errors.Is(err, collections.ErrNotFound) { + return nil, status.Errorf(codes.Internal, "parent UTX %s not found for pending outbound %s", entry.UniversalTxId, req.OutboundId) + } + return nil, status.Error(codes.Internal, err.Error()) + } + + // Find the outbound in utx.OutboundTx[] by ID + var outbound *types.OutboundTx + for _, ob := range utx.OutboundTx { + if ob != nil && ob.Id == req.OutboundId { + outbound = ob + break + } + } + + return &types.QueryGetPendingOutboundResponse{ + Entry: &entry, + Outbound: outbound, + }, nil +} + +// AllPendingOutbounds implements types.QueryServer. +// Returns a paginated list of all pending outbound entries with full outbound data. +func (k Querier) AllPendingOutbounds(goCtx context.Context, req *types.QueryAllPendingOutboundsRequest) (*types.QueryAllPendingOutboundsResponse, error) { + if req == nil { + return nil, status.Error(codes.InvalidArgument, "invalid request") + } + + ctx := sdk.UnwrapSDKContext(goCtx) + + entries, pageRes, err := query.CollectionPaginate(ctx, k.PendingOutbounds, req.Pagination, func(_ string, value types.PendingOutboundEntry) (*types.PendingOutboundEntry, error) { + return &value, nil + }) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + + // Resolve full outbound data from parent UTXs + var outbounds []*types.OutboundTx + for _, entry := range entries { + utx, err := k.UniversalTx.Get(ctx, entry.UniversalTxId) + if err != nil { + continue // skip if UTX not found + } + for _, ob := range utx.OutboundTx { + if ob != nil && ob.Id == entry.OutboundId { + outbounds = append(outbounds, ob) + break + } + } + } + + return &types.QueryAllPendingOutboundsResponse{ + Entries: entries, + Outbounds: outbounds, + Pagination: pageRes, + }, nil +} + // chainMetaToGasPrice converts a ChainMeta into the legacy GasPrice shape // so that existing API consumers see no breaking change. func chainMetaToGasPrice(cm *types.ChainMeta) *types.GasPrice { diff --git a/x/uexecutor/module.go b/x/uexecutor/module.go index dbe947e8e..1fe9b66d3 100755 --- a/x/uexecutor/module.go +++ b/x/uexecutor/module.go @@ -30,7 +30,8 @@ import ( const ( // ConsensusVersion defines the current x/uexecutor module consensus version. - ConsensusVersion = 5 + // Bumped to 6: added PendingOutbounds collection. + ConsensusVersion = 6 ) var ( @@ -190,6 +191,14 @@ func (a AppModule) RegisterServices(cfg module.Configurator) { if err := cfg.RegisterMigration(types.ModuleName, 4, a.migrateToV5()); err != nil { panic(fmt.Sprintf("failed to migrate %s from version 4 to 5: %v", types.ModuleName, err)) } + + // Register migration from version 5 -> 6 (pending-outbounds-index) + // New PendingOutbounds collection starts empty; the upgrade handler backfills it. + if err := cfg.RegisterMigration(types.ModuleName, 5, func(ctx sdk.Context) error { + return nil // no-op: new collection initialized empty by schema builder + }); err != nil { + panic(fmt.Sprintf("failed to migrate %s from version 5 to 6: %v", types.ModuleName, err)) + } } func (a AppModule) migrateToV2() module.MigrationHandler { diff --git a/x/uexecutor/types/keys.go b/x/uexecutor/types/keys.go index 9716aa38d..615f324e9 100755 --- a/x/uexecutor/types/keys.go +++ b/x/uexecutor/types/keys.go @@ -33,6 +33,9 @@ var ( ChainMetaKey = collections.NewPrefix(6) ChainMetasName = "chain_metas" + + PendingOutboundsKey = collections.NewPrefix(7) + PendingOutboundsName = "pending_outbounds" ) const ( diff --git a/x/uexecutor/types/pending_outbound.pb.go b/x/uexecutor/types/pending_outbound.pb.go new file mode 100644 index 000000000..ca8593b07 --- /dev/null +++ b/x/uexecutor/types/pending_outbound.pb.go @@ -0,0 +1,429 @@ +// This file contains the Go types for the PendingOutbound protobuf messages. +// These types will be replaced by protoc-gen-gogo output when `make proto-gen` +// is run. They are provided here so that the rest of the codebase compiles +// before code generation is executed. + +package types + +import ( + fmt "fmt" + io "io" + math "math" + math_bits "math/bits" + + query "github.com/cosmos/cosmos-sdk/types/query" + proto "github.com/cosmos/gogoproto/proto" +) + +// Ensure proto imports are used. +var _ = fmt.Errorf +var _ = math.Inf + +// PendingOutboundEntry is a lightweight index entry for a pending outbound. +type PendingOutboundEntry struct { + OutboundId string `protobuf:"bytes,1,opt,name=outbound_id,json=outboundId,proto3" json:"outbound_id,omitempty"` + UniversalTxId string `protobuf:"bytes,2,opt,name=universal_tx_id,json=universalTxId,proto3" json:"universal_tx_id,omitempty"` + CreatedAt int64 `protobuf:"varint,3,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"` +} + +func (m *PendingOutboundEntry) Reset() { *m = PendingOutboundEntry{} } +func (m *PendingOutboundEntry) String() string { return proto.CompactTextString(m) } +func (*PendingOutboundEntry) ProtoMessage() {} + +func (m *PendingOutboundEntry) GetOutboundId() string { + if m != nil { + return m.OutboundId + } + return "" +} + +func (m *PendingOutboundEntry) GetUniversalTxId() string { + if m != nil { + return m.UniversalTxId + } + return "" +} + +func (m *PendingOutboundEntry) GetCreatedAt() int64 { + if m != nil { + return m.CreatedAt + } + return 0 +} + +func (m *PendingOutboundEntry) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *PendingOutboundEntry) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *PendingOutboundEntry) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + if m.CreatedAt != 0 { + i = encodeVarintPendingOutbound(dAtA, i, uint64(m.CreatedAt)) + i-- + dAtA[i] = 0x18 + } + if len(m.UniversalTxId) > 0 { + i -= len(m.UniversalTxId) + copy(dAtA[i:], m.UniversalTxId) + i = encodeVarintPendingOutbound(dAtA, i, uint64(len(m.UniversalTxId))) + i-- + dAtA[i] = 0x12 + } + if len(m.OutboundId) > 0 { + i -= len(m.OutboundId) + copy(dAtA[i:], m.OutboundId) + i = encodeVarintPendingOutbound(dAtA, i, uint64(len(m.OutboundId))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func encodeVarintPendingOutbound(dAtA []byte, offset int, v uint64) int { + offset-- + dAtA[offset] = uint8(v) + for v >= 1<<7 { + dAtA[offset] = uint8(v&0x7f | 0x80) + v >>= 7 + offset-- + } + dAtA[offset] = uint8(v) + return offset +} + +func (m *PendingOutboundEntry) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.OutboundId) + if l > 0 { + n += 1 + l + sovPendingOutbound(uint64(l)) + } + l = len(m.UniversalTxId) + if l > 0 { + n += 1 + l + sovPendingOutbound(uint64(l)) + } + if m.CreatedAt != 0 { + n += 1 + sovPendingOutbound(uint64(m.CreatedAt)) + } + return n +} + +func sovPendingOutbound(x uint64) (n int) { + return (math_bits.Len64(x|1) + 6) / 7 +} + +func (m *PendingOutboundEntry) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return fmt.Errorf("proto: integer overflow") + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + switch fieldNum { + case 1: // outbound_id + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field OutboundId", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return fmt.Errorf("proto: integer overflow") + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return fmt.Errorf("proto: negative length found during unmarshaling") + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return fmt.Errorf("proto: negative length found during unmarshaling") + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.OutboundId = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: // universal_tx_id + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field UniversalTxId", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return fmt.Errorf("proto: integer overflow") + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return fmt.Errorf("proto: negative length found during unmarshaling") + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return fmt.Errorf("proto: negative length found during unmarshaling") + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.UniversalTxId = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 3: // created_at + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field CreatedAt", wireType) + } + m.CreatedAt = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return fmt.Errorf("proto: integer overflow") + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.CreatedAt |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + default: + iNdEx = preIndex + skippy, err := skipPendingOutbound(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return fmt.Errorf("proto: negative length found during unmarshaling") + } + if iNdEx+skippy > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} + +func skipPendingOutbound(dAtA []byte) (n int, err error) { + l := len(dAtA) + iNdEx := 0 + depth := 0 + for iNdEx < l { + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, fmt.Errorf("proto: integer overflow") + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + wireType := int(wire & 0x7) + switch wireType { + case 0: + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, fmt.Errorf("proto: integer overflow") + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + iNdEx++ + if dAtA[iNdEx-1] < 0x80 { + break + } + } + case 1: + iNdEx += 8 + case 2: + var length int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, fmt.Errorf("proto: integer overflow") + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + length |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if length < 0 { + return 0, fmt.Errorf("proto: negative length found during unmarshaling") + } + iNdEx += length + case 3: + depth++ + case 4: + if depth == 0 { + return 0, fmt.Errorf("proto: unexpected end of group") + } + depth-- + case 5: + iNdEx += 4 + default: + return 0, fmt.Errorf("proto: illegal wireType %d", wireType) + } + if iNdEx < 0 { + return 0, fmt.Errorf("proto: negative length found during unmarshaling") + } + if depth == 0 { + return iNdEx, nil + } + } + return 0, io.ErrUnexpectedEOF +} + +// Query request/response types for pending outbounds + +type QueryGetPendingOutboundRequest struct { + OutboundId string `protobuf:"bytes,1,opt,name=outbound_id,json=outboundId,proto3" json:"outbound_id,omitempty"` +} + +func (m *QueryGetPendingOutboundRequest) Reset() { *m = QueryGetPendingOutboundRequest{} } +func (m *QueryGetPendingOutboundRequest) String() string { return proto.CompactTextString(m) } +func (*QueryGetPendingOutboundRequest) ProtoMessage() {} + +func (m *QueryGetPendingOutboundRequest) GetOutboundId() string { + if m != nil { + return m.OutboundId + } + return "" +} + +type QueryGetPendingOutboundResponse struct { + Entry *PendingOutboundEntry `protobuf:"bytes,1,opt,name=entry,proto3" json:"entry,omitempty"` + Outbound *OutboundTx `protobuf:"bytes,2,opt,name=outbound,proto3" json:"outbound,omitempty"` +} + +func (m *QueryGetPendingOutboundResponse) Reset() { *m = QueryGetPendingOutboundResponse{} } +func (m *QueryGetPendingOutboundResponse) String() string { return proto.CompactTextString(m) } +func (*QueryGetPendingOutboundResponse) ProtoMessage() {} + +func (m *QueryGetPendingOutboundResponse) GetEntry() *PendingOutboundEntry { + if m != nil { + return m.Entry + } + return nil +} + +func (m *QueryGetPendingOutboundResponse) GetOutbound() *OutboundTx { + if m != nil { + return m.Outbound + } + return nil +} + +type QueryAllPendingOutboundsRequest struct { + Pagination *query.PageRequest `protobuf:"bytes,1,opt,name=pagination,proto3" json:"pagination,omitempty"` +} + +func (m *QueryAllPendingOutboundsRequest) Reset() { *m = QueryAllPendingOutboundsRequest{} } +func (m *QueryAllPendingOutboundsRequest) String() string { return proto.CompactTextString(m) } +func (*QueryAllPendingOutboundsRequest) ProtoMessage() {} + +func (m *QueryAllPendingOutboundsRequest) GetPagination() *query.PageRequest { + if m != nil { + return m.Pagination + } + return nil +} + +type QueryAllPendingOutboundsResponse struct { + Entries []*PendingOutboundEntry `protobuf:"bytes,1,rep,name=entries,proto3" json:"entries,omitempty"` + Outbounds []*OutboundTx `protobuf:"bytes,2,rep,name=outbounds,proto3" json:"outbounds,omitempty"` + Pagination *query.PageResponse `protobuf:"bytes,3,opt,name=pagination,proto3" json:"pagination,omitempty"` +} + +func (m *QueryAllPendingOutboundsResponse) Reset() { *m = QueryAllPendingOutboundsResponse{} } +func (m *QueryAllPendingOutboundsResponse) String() string { return proto.CompactTextString(m) } +func (*QueryAllPendingOutboundsResponse) ProtoMessage() {} + +func (m *QueryAllPendingOutboundsResponse) GetOutbounds() []*OutboundTx { + if m != nil { + return m.Outbounds + } + return nil +} + +func (m *QueryAllPendingOutboundsResponse) GetEntries() []*PendingOutboundEntry { + if m != nil { + return m.Entries + } + return nil +} + +func (m *QueryAllPendingOutboundsResponse) GetPagination() *query.PageResponse { + if m != nil { + return m.Pagination + } + return nil +} + +func init() { + proto.RegisterType((*PendingOutboundEntry)(nil), "uexecutor.v1.PendingOutboundEntry") + proto.RegisterType((*QueryGetPendingOutboundRequest)(nil), "uexecutor.v1.QueryGetPendingOutboundRequest") + proto.RegisterType((*QueryGetPendingOutboundResponse)(nil), "uexecutor.v1.QueryGetPendingOutboundResponse") + proto.RegisterType((*QueryAllPendingOutboundsRequest)(nil), "uexecutor.v1.QueryAllPendingOutboundsRequest") + proto.RegisterType((*QueryAllPendingOutboundsResponse)(nil), "uexecutor.v1.QueryAllPendingOutboundsResponse") +} diff --git a/x/uexecutor/types/query.pb.go b/x/uexecutor/types/query.pb.go index d47a8f5cb..be4c1e2e0 100644 --- a/x/uexecutor/types/query.pb.go +++ b/x/uexecutor/types/query.pb.go @@ -973,6 +973,10 @@ type QueryServer interface { ChainMeta(context.Context, *QueryChainMetaRequest) (*QueryChainMetaResponse, error) // 🔹 Queries all chain metas across chains AllChainMetas(context.Context, *QueryAllChainMetasRequest) (*QueryAllChainMetasResponse, error) + // Get a single pending outbound by ID + GetPendingOutbound(context.Context, *QueryGetPendingOutboundRequest) (*QueryGetPendingOutboundResponse, error) + // Get all pending outbounds (paginated) + AllPendingOutbounds(context.Context, *QueryAllPendingOutboundsRequest) (*QueryAllPendingOutboundsResponse, error) } // UnimplementedQueryServer can be embedded to have forward compatible implementations. @@ -1003,6 +1007,12 @@ func (*UnimplementedQueryServer) ChainMeta(ctx context.Context, req *QueryChainM func (*UnimplementedQueryServer) AllChainMetas(ctx context.Context, req *QueryAllChainMetasRequest) (*QueryAllChainMetasResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method AllChainMetas not implemented") } +func (*UnimplementedQueryServer) GetPendingOutbound(ctx context.Context, req *QueryGetPendingOutboundRequest) (*QueryGetPendingOutboundResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetPendingOutbound not implemented") +} +func (*UnimplementedQueryServer) AllPendingOutbounds(ctx context.Context, req *QueryAllPendingOutboundsRequest) (*QueryAllPendingOutboundsResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method AllPendingOutbounds not implemented") +} func RegisterQueryServer(s grpc1.Server, srv QueryServer) { s.RegisterService(&_Query_serviceDesc, srv) @@ -1152,6 +1162,42 @@ func _Query_AllChainMetas_Handler(srv interface{}, ctx context.Context, dec func return interceptor(ctx, in, info, handler) } +func _Query_GetPendingOutbound_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(QueryGetPendingOutboundRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(QueryServer).GetPendingOutbound(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/uexecutor.v1.Query/GetPendingOutbound", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(QueryServer).GetPendingOutbound(ctx, req.(*QueryGetPendingOutboundRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Query_AllPendingOutbounds_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(QueryAllPendingOutboundsRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(QueryServer).AllPendingOutbounds(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/uexecutor.v1.Query/AllPendingOutbounds", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(QueryServer).AllPendingOutbounds(ctx, req.(*QueryAllPendingOutboundsRequest)) + } + return interceptor(ctx, in, info, handler) +} + var _Query_serviceDesc = grpc.ServiceDesc{ ServiceName: "uexecutor.v1.Query", HandlerType: (*QueryServer)(nil), @@ -1188,6 +1234,14 @@ var _Query_serviceDesc = grpc.ServiceDesc{ MethodName: "AllChainMetas", Handler: _Query_AllChainMetas_Handler, }, + { + MethodName: "GetPendingOutbound", + Handler: _Query_GetPendingOutbound_Handler, + }, + { + MethodName: "AllPendingOutbounds", + Handler: _Query_AllPendingOutbounds_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "uexecutor/v1/query.proto",