diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index ccf5aca0..4a2393c8 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -139,19 +139,45 @@ func retryWithRoundRobin[T any]( // GetAllChainConfigs retrieves all chain configurations from Push Chain. func (c *Client) GetAllChainConfigs(ctx context.Context) ([]*uregistrytypes.ChainConfig, error) { - return retryWithRoundRobin( - len(c.eps), - &c.rr, - func(idx int) ([]*uregistrytypes.ChainConfig, error) { - resp, err := c.eps[idx].AllChainConfigs(ctx, &uregistrytypes.QueryAllChainConfigsRequest{}) - if err != nil { - return nil, err - } - return resp.Configs, nil - }, - "GetAllChainConfigs", - c.logger, + // Paged rather than a single request: the server paginates this collection, + // and an omitted PageRequest silently caps the response at the SDK default of + // 100. A chain missing from this list is simply never watched, so truncation + // must not be possible. + var ( + configs []*uregistrytypes.ChainConfig + nextKey []byte ) + for page := 0; page < chainConfigMaxPages; page++ { + key := nextKey + resp, err := retryWithRoundRobin( + len(c.eps), + &c.rr, + func(idx int) (*uregistrytypes.QueryAllChainConfigsResponse, error) { + return c.eps[idx].AllChainConfigs(ctx, &uregistrytypes.QueryAllChainConfigsRequest{ + Pagination: &query.PageRequest{Key: key, Limit: chainConfigPageSize}, + }) + }, + "GetAllChainConfigs", + c.logger, + ) + if err != nil { + return nil, err + } + + configs = append(configs, resp.Configs...) + + if resp.Pagination == nil || len(resp.Pagination.NextKey) == 0 { + return configs, nil + } + nextKey = resp.Pagination.NextKey + } + + // Unreachable with any plausible number of chains; loud rather than silent. + c.logger.Error(). + Int("max_pages", chainConfigMaxPages). + Int("fetched", len(configs)). + Msg("chain config page cap reached; some chains will not be watched") + return configs, nil } // GetLatestBlock retrieves the latest block from Push Chain. @@ -368,15 +394,44 @@ func (c *Client) GetPendingFundMigrations(ctx context.Context) ([]*utsstypes.Fun ) } -// GetAllPendingOutbounds retrieves up to the first 1000 pending outbound transactions from Push Chain. -// Sorted by created_at (block height) ascending — oldest first. +// Page size and page cap for the pending-outbound walk. The cap bounds a single +// poll; the remainder is picked up on the next tick. +const ( + // AllPendingOutbounds pages by offset and returns only Total, never a NextKey, + // so a key-based walk stops after one page. It also loads and sorts the whole + // collection per call, so paging saves the server nothing. The sort is not + // stable and orders on CreatedAt, a block height, so rows sharing a height can + // change relative order between calls — which makes offset paging able to skip + // a row outright. One generous request plus a Total check is the only shape + // that is both correct and cheap against that server. + // A row costs roughly a kilobyte on the wire, so this stays well inside gRPC's + // 4 MiB default. Asking for the whole set instead would fail the call outright + // once the set grew, taking the poll down rather than returning a short list. + pendingOutboundLimit = 1000 + + chainConfigPageSize = 200 + chainConfigMaxPages = 20 +) + +// GetAllPendingOutbounds retrieves pending outbound transactions from Push Chain. +// +// Read newest-first. An outbound only leaves the pending set once a quorum vote +// terminalizes it, so a row that cannot reach one stays at the head of an +// oldest-first list forever and would hide every newer outbound behind it — on +// every chain, since this query is not chain-scoped. New outbounds always arrive +// at the newest end, so reading that end cannot be starved. +// +// This is discovery only and does not set signing priority. The event store +// hands work to the signer ordered by block_height ASC, so older outbounds are +// still signed first; reading newest-first only decides what reaches the store +// to be ordered in the first place. func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes.PendingOutboundEntry, []*uexecutortypes.OutboundTx, error) { resp, err := retryWithRoundRobin( len(c.uexecutorClients), &c.rr, func(idx int) (*uexecutortypes.QueryAllPendingOutboundsResponse, error) { return c.uexecutorClients[idx].AllPendingOutbounds(ctx, &uexecutortypes.QueryAllPendingOutboundsRequest{ - Pagination: &query.PageRequest{Limit: 1000}, + Pagination: &query.PageRequest{Limit: pendingOutboundLimit, Reverse: true}, }) }, "GetAllPendingOutbounds", @@ -385,6 +440,18 @@ func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes. if err != nil { return nil, nil, err } + + // Below the limit this read is the whole set and the direction is irrelevant. + // Above it, the direction is the point: anything older is already known + // locally, so continuing to re-read it would achieve nothing while the newer + // rows went unsigned. + if resp.Pagination != nil && resp.Pagination.Total > pendingOutboundLimit { + c.logger.Warn(). + Uint64("total", resp.Pagination.Total). + Uint64("limit", pendingOutboundLimit). + Msg("pending outbound set exceeds one request; only the newest are read, older rows must already be known locally") + } + return resp.Entries, resp.Outbounds, nil } diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index e7b88e04..cbcd6fb4 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -1,6 +1,7 @@ package pushcore import ( + "bytes" "context" "errors" "math/big" @@ -8,6 +9,7 @@ import ( cmtservice "github.com/cosmos/cosmos-sdk/client/grpc/cmtservice" sdktypes "github.com/cosmos/cosmos-sdk/types" + "github.com/cosmos/cosmos-sdk/types/query" "github.com/cosmos/cosmos-sdk/types/tx" authtypes "github.com/cosmos/cosmos-sdk/x/auth/types" "github.com/cosmos/cosmos-sdk/x/authz" @@ -917,9 +919,24 @@ type mockRegistryQueryClient struct { uregistrytypes.QueryClient allChainConfigsResp *uregistrytypes.QueryAllChainConfigsResponse err error + + chainConfigPages []*uregistrytypes.QueryAllChainConfigsResponse + chainConfigKeys [][]byte } func (m *mockRegistryQueryClient) AllChainConfigs(ctx context.Context, req *uregistrytypes.QueryAllChainConfigsRequest, opts ...grpc.CallOption) (*uregistrytypes.QueryAllChainConfigsResponse, error) { + if m.chainConfigPages != nil { + var key []byte + if req.Pagination != nil { + key = req.Pagination.Key + } + m.chainConfigKeys = append(m.chainConfigKeys, key) + idx := len(m.chainConfigKeys) - 1 + if idx >= len(m.chainConfigPages) { + return nil, assert.AnError + } + return m.chainConfigPages[idx], nil + } if m.err != nil { return nil, m.err } @@ -1028,6 +1045,9 @@ type mockUExecutorQueryClient struct { gasPriceResp *uexecutortypes.QueryGasPriceResponse allPendingOutboundsResp *uexecutortypes.QueryAllPendingOutboundsResponse err error + + // lastPendingReq records the request so tests can assert the limit sent. + lastPendingReq *uexecutortypes.QueryAllPendingOutboundsRequest } func (m *mockUExecutorQueryClient) GasPrice(ctx context.Context, req *uexecutortypes.QueryGasPriceRequest, opts ...grpc.CallOption) (*uexecutortypes.QueryGasPriceResponse, error) { @@ -1054,6 +1074,7 @@ func (m *mockUExecutorQueryClient) AllUniversalTx(ctx context.Context, req *uexe } func (m *mockUExecutorQueryClient) AllPendingOutbounds(ctx context.Context, req *uexecutortypes.QueryAllPendingOutboundsRequest, opts ...grpc.CallOption) (*uexecutortypes.QueryAllPendingOutboundsResponse, error) { + m.lastPendingReq = req if m.err != nil { return nil, m.err } @@ -1149,3 +1170,55 @@ func TestClient_GetKeyByID(t *testing.T) { assert.Nil(t, key) }) } + +// An outbound only leaves the pending set on a quorum vote, so one that cannot +// reach a vote sits at the head of an oldest-first list permanently and hides +// everything newer. New outbounds always arrive at the newest end, so reading +// that end is what cannot be starved. +func TestClient_GetAllPendingOutbounds_ReadsNewestFirst(t *testing.T) { + ctx := context.Background() + + resp := func(total uint64) *uexecutortypes.QueryAllPendingOutboundsResponse { + return &uexecutortypes.QueryAllPendingOutboundsResponse{ + Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: "ob-1"}}, + Outbounds: []*uexecutortypes.OutboundTx{{Id: "ob-1"}}, + Pagination: &query.PageResponse{Total: total}, + } + } + + t.Run("reads the newest end, never an offset", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(9)} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + _, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + + p := mockClient.lastPendingReq.Pagination + require.NotNil(t, p) + assert.True(t, p.Reverse, "a stuck prefix at the oldest end must not hide newer rows") + assert.Zero(t, p.Offset, "offset zero is the only position that cannot shift under insertion") + assert.Equal(t, uint64(pendingOutboundLimit), p.Limit) + }) + + // Above the limit older rows stop being read. They are already known locally, + // but an operator should still be told the set is that large. + t.Run("reports a set larger than one request", func(t *testing.T) { + var logBuf bytes.Buffer + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(pendingOutboundLimit + 500)} + client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + _, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + assert.Contains(t, logBuf.String(), "exceeds one request") + }) + + t.Run("quiet when the set fits", func(t *testing.T) { + var logBuf bytes.Buffer + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(9)} + client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + _, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + assert.NotContains(t, logBuf.String(), "exceeds one request") + }) +}