From 8a7c744cb390675b38003143c281c1203f2c147a Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 16:47:47 +0530 Subject: [PATCH 1/8] fix: paginate pending outbound polling so a stuck prefix cannot hide newer rows (F-2026-18817) --- universalClient/pushcore/pushCore.go | 72 +++++++++--- universalClient/pushcore/pushCore_test.go | 134 ++++++++++++++++++++++ 2 files changed, 191 insertions(+), 15 deletions(-) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index ccf5aca0..d28545c3 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -368,24 +368,66 @@ 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 ( + pendingOutboundPageSize = 1000 + pendingOutboundMaxPages = 20 +) + +// GetAllPendingOutbounds retrieves pending outbound transactions from Push Chain, +// sorted by created_at (block height) ascending — oldest first. +// +// The result is paged rather than a single query. An outbound only leaves the +// pending set once a quorum vote terminalizes it, so any row that cannot reach +// one — for example a destination execution whose observation was never seen — +// stays at the head of an oldest-first list forever. Reading one fixed page +// would let such a prefix hide every newer outbound from signing, on every +// chain, since this query is not chain-scoped. 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}, - }) - }, - "GetAllPendingOutbounds", - c.logger, + var ( + entries []*uexecutortypes.PendingOutboundEntry + outbounds []*uexecutortypes.OutboundTx + nextKey []byte ) - if err != nil { - return nil, nil, err + + for page := 0; page < pendingOutboundMaxPages; page++ { + key := nextKey + 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{Key: key, Limit: pendingOutboundPageSize}, + }) + }, + "GetAllPendingOutbounds", + c.logger, + ) + if err != nil { + // Return what we have rather than nothing: a later page failing must + // not stop the caller acting on the pages that did arrive. + if len(entries) > 0 { + c.logger.Warn().Err(err).Int("page", page).Msg("pending outbound page failed, using pages fetched so far") + return entries, outbounds, nil + } + return nil, nil, err + } + + entries = append(entries, resp.Entries...) + outbounds = append(outbounds, resp.Outbounds...) + + if resp.Pagination == nil || len(resp.Pagination.NextKey) == 0 { + return entries, outbounds, nil + } + nextKey = resp.Pagination.NextKey } - return resp.Entries, resp.Outbounds, nil + + c.logger.Warn(). + Int("max_pages", pendingOutboundMaxPages). + Int("fetched", len(entries)). + Msg("pending outbound page cap reached; remainder deferred to next poll") + return entries, outbounds, nil } // createGRPCConnection creates a gRPC connection with appropriate transport security. diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index e7b88e04..544114a6 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -3,11 +3,13 @@ package pushcore import ( "context" "errors" + "fmt" "math/big" "testing" 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" @@ -1028,6 +1030,12 @@ type mockUExecutorQueryClient struct { gasPriceResp *uexecutortypes.QueryGasPriceResponse allPendingOutboundsResp *uexecutortypes.QueryAllPendingOutboundsResponse err error + + // Paging support: `pages` is served in order, one per call, and the key the + // caller sent is recorded so tests can assert NextKey is actually followed. + pages []*uexecutortypes.QueryAllPendingOutboundsResponse + requestedKeys [][]byte + failAfterPage int // when > 0, calls beyond this page number return err } func (m *mockUExecutorQueryClient) GasPrice(ctx context.Context, req *uexecutortypes.QueryGasPriceRequest, opts ...grpc.CallOption) (*uexecutortypes.QueryGasPriceResponse, error) { @@ -1054,6 +1062,21 @@ 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) { + if m.pages != nil { + var key []byte + if req.Pagination != nil { + key = req.Pagination.Key + } + m.requestedKeys = append(m.requestedKeys, key) + idx := len(m.requestedKeys) - 1 + if m.failAfterPage > 0 && idx >= m.failAfterPage { + return nil, assert.AnError + } + if idx >= len(m.pages) { + return nil, assert.AnError + } + return m.pages[idx], nil + } if m.err != nil { return nil, m.err } @@ -1149,3 +1172,114 @@ func TestClient_GetKeyByID(t *testing.T) { assert.Nil(t, key) }) } + +// page builds one response carrying a single outbound, with nextKey signalling +// whether more pages follow. +func pendingPage(id string, nextKey []byte) *uexecutortypes.QueryAllPendingOutboundsResponse { + return &uexecutortypes.QueryAllPendingOutboundsResponse{ + Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: id, UniversalTxId: "utx-" + id}}, + Outbounds: []*uexecutortypes.OutboundTx{{Id: id, DestinationChain: "eip155:1"}}, + Pagination: &query.PageResponse{NextKey: nextKey}, + } +} + +// Pending outbounds are oldest-first and only leave the set on a quorum vote, so +// a row that can never reach one sits at the head permanently. Reading a single +// page would hide every newer outbound behind it, on every chain. +func TestClient_GetAllPendingOutbounds_Paginates(t *testing.T) { + ctx := context.Background() + + t.Run("follows NextKey until exhausted", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ + pendingPage("stuck", []byte("k1")), + pendingPage("also-stuck", []byte("k2")), + pendingPage("reachable", nil), // last page + }} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + entries, outbounds, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + require.Len(t, entries, 3) + require.Len(t, outbounds, 3) + + // The outbound behind the stuck prefix must be visible. + assert.Equal(t, "reachable", entries[2].OutboundId) + + // And the cursor must actually be threaded through, not just re-queried. + assert.Equal(t, [][]byte{nil, []byte("k1"), []byte("k2")}, mockClient.requestedKeys) + }) + + t.Run("single page with no NextKey stops immediately", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ + pendingPage("only", nil), + }} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + entries, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + require.Len(t, entries, 1) + assert.Len(t, mockClient.requestedKeys, 1, "must not request a second page") + }) + + t.Run("nil pagination stops the walk", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ + {Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: "ob-1"}}, + Outbounds: []*uexecutortypes.OutboundTx{{Id: "ob-1"}}}, + }} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + entries, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + require.Len(t, entries, 1) + assert.Len(t, mockClient.requestedKeys, 1) + }) + + // A later page failing must not discard the earlier ones: those outbounds are + // still signable, and returning nothing would stall them for no reason. + t.Run("mid-walk failure keeps the pages already fetched", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{ + pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ + pendingPage("ob-1", []byte("k1")), + pendingPage("ob-2", []byte("k2")), + }, + failAfterPage: 1, + } + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + entries, outbounds, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + require.Len(t, entries, 1) + require.Len(t, outbounds, 1) + assert.Equal(t, "ob-1", entries[0].OutboundId) + }) + + t.Run("first page failing is an error", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{ + pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{pendingPage("ob-1", nil)}, + failAfterPage: 0, + err: assert.AnError, + } + mockClient.pages = nil // fall through to the plain error path + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + entries, _, err := client.GetAllPendingOutbounds(ctx) + require.Error(t, err) + assert.Nil(t, entries) + }) + + // The cap bounds one poll; the rest is picked up next tick rather than + // growing the request without limit. + t.Run("stops at the page cap", func(t *testing.T) { + pages := make([]*uexecutortypes.QueryAllPendingOutboundsResponse, pendingOutboundMaxPages+5) + for i := range pages { + pages[i] = pendingPage(fmt.Sprintf("ob-%d", i), []byte("more")) + } + mockClient := &mockUExecutorQueryClient{pages: pages} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + entries, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + assert.Len(t, entries, pendingOutboundMaxPages) + assert.Len(t, mockClient.requestedKeys, pendingOutboundMaxPages) + }) +} From 8ca34bd1ee91beb3d7f5dc1f718ba1a8e96765da Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 17:09:39 +0530 Subject: [PATCH 2/8] fix: paginate chain config query so it cannot silently cap at the SDK default (F-2026-18817) --- universalClient/pushcore/pushCore.go | 53 ++++++++++++++----- universalClient/pushcore/pushCore_test.go | 64 +++++++++++++++++++++++ 2 files changed, 105 insertions(+), 12 deletions(-) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index d28545c3..1cc43bdd 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. @@ -373,6 +399,9 @@ func (c *Client) GetPendingFundMigrations(ctx context.Context) ([]*utsstypes.Fun const ( pendingOutboundPageSize = 1000 pendingOutboundMaxPages = 20 + + chainConfigPageSize = 200 + chainConfigMaxPages = 20 ) // GetAllPendingOutbounds retrieves pending outbound transactions from Push Chain, diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index 544114a6..3f70ba95 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -919,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 } @@ -1283,3 +1298,52 @@ func TestClient_GetAllPendingOutbounds_Paginates(t *testing.T) { assert.Len(t, mockClient.requestedKeys, pendingOutboundMaxPages) }) } + +func chainConfigPage(chain string, nextKey []byte) *uregistrytypes.QueryAllChainConfigsResponse { + return &uregistrytypes.QueryAllChainConfigsResponse{ + Configs: []*uregistrytypes.ChainConfig{{Chain: chain}}, + Pagination: &query.PageResponse{NextKey: nextKey}, + } +} + +// The server paginates this collection, so omitting a PageRequest capped the +// response at the SDK default of 100. A chain missing from the list is simply +// never watched, so the walk must not be able to truncate. +func TestClient_GetAllChainConfigs_Paginates(t *testing.T) { + ctx := context.Background() + + t.Run("follows NextKey across pages", func(t *testing.T) { + mockClient := &mockRegistryQueryClient{chainConfigPages: []*uregistrytypes.QueryAllChainConfigsResponse{ + chainConfigPage("eip155:1", []byte("k1")), + chainConfigPage("solana:x", nil), + }} + client := &Client{logger: zerolog.Nop(), eps: []uregistrytypes.QueryClient{mockClient}} + + configs, err := client.GetAllChainConfigs(ctx) + require.NoError(t, err) + require.Len(t, configs, 2) + assert.Equal(t, "solana:x", configs[1].Chain) + assert.Equal(t, [][]byte{nil, []byte("k1")}, mockClient.chainConfigKeys) + }) + + t.Run("a page request is always sent", func(t *testing.T) { + mockClient := &mockRegistryQueryClient{chainConfigPages: []*uregistrytypes.QueryAllChainConfigsResponse{ + chainConfigPage("eip155:1", nil), + }} + client := &Client{logger: zerolog.Nop(), eps: []uregistrytypes.QueryClient{mockClient}} + + _, err := client.GetAllChainConfigs(ctx) + require.NoError(t, err) + require.Len(t, mockClient.chainConfigKeys, 1, "must not request a second page") + }) + + t.Run("error propagates", func(t *testing.T) { + client := &Client{logger: zerolog.Nop(), eps: []uregistrytypes.QueryClient{ + &mockRegistryQueryClient{err: assert.AnError}, + }} + + configs, err := client.GetAllChainConfigs(ctx) + require.Error(t, err) + assert.Nil(t, configs) + }) +} From 19c3b6ff2d085c54a3bb5f753afb07e9dcfbfcfe Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 17:22:18 +0530 Subject: [PATCH 3/8] fix: carry pending outbound cursor between polls so the page budget costs latency not coverage (F-2026-18817) --- universalClient/pushcore/pushCore.go | 26 +++++++++-- universalClient/pushcore/pushCore_test.go | 55 +++++++++++++++++++++++ 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index 1cc43bdd..d82280b5 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -9,6 +9,7 @@ import ( "fmt" "math/big" "strings" + "sync" "sync/atomic" cmtservice "github.com/cosmos/cosmos-sdk/client/grpc/cmtservice" @@ -40,6 +41,13 @@ type Client struct { authClients []authtypes.QueryClient // Auth query clients conns []*grpc.ClientConn // Owned gRPC connections (for cleanup) rr uint32 // Round-robin counter for endpoint selection + + // Pagination cursor carried between pending-outbound polls. The page budget + // bounds one poll's work; this makes the budget cost latency rather than + // coverage, so a set larger than the budget is still walked in full over + // successive ticks. Nil means start from the beginning. + pendingMu sync.Mutex + pendingCursor []byte } // New creates a new Client by dialing the provided gRPC URLs. @@ -414,10 +422,13 @@ const ( // would let such a prefix hide every newer outbound from signing, on every // chain, since this query is not chain-scoped. func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes.PendingOutboundEntry, []*uexecutortypes.OutboundTx, error) { + c.pendingMu.Lock() + defer c.pendingMu.Unlock() + var ( entries []*uexecutortypes.PendingOutboundEntry outbounds []*uexecutortypes.OutboundTx - nextKey []byte + nextKey = c.pendingCursor ) for page := 0; page < pendingOutboundMaxPages; page++ { @@ -435,11 +446,14 @@ func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes. ) if err != nil { // Return what we have rather than nothing: a later page failing must - // not stop the caller acting on the pages that did arrive. + // not stop the caller acting on the pages that did arrive. Resume from + // the failed page next time instead of losing the ground already made. if len(entries) > 0 { + c.pendingCursor = key c.logger.Warn().Err(err).Int("page", page).Msg("pending outbound page failed, using pages fetched so far") return entries, outbounds, nil } + c.pendingCursor = nil return nil, nil, err } @@ -447,15 +461,21 @@ func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes. outbounds = append(outbounds, resp.Outbounds...) if resp.Pagination == nil || len(resp.Pagination.NextKey) == 0 { + // Reached the end; the next poll starts from the beginning again so + // rows added at the tail since the walk began are picked up. + c.pendingCursor = nil return entries, outbounds, nil } nextKey = resp.Pagination.NextKey } + // Budget spent mid-set. Park the cursor so the next tick continues from here + // rather than re-reading the same prefix forever. + c.pendingCursor = nextKey c.logger.Warn(). Int("max_pages", pendingOutboundMaxPages). Int("fetched", len(entries)). - Msg("pending outbound page cap reached; remainder deferred to next poll") + Msg("pending outbound page budget spent; continuing from this cursor next poll") return entries, outbounds, nil } diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index 3f70ba95..9fcb1439 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -1347,3 +1347,58 @@ func TestClient_GetAllChainConfigs_Paginates(t *testing.T) { assert.Nil(t, configs) }) } + +// The page budget bounds one poll's work, so it must cost latency rather than +// coverage: a set larger than the budget has to be walked in full across +// successive polls instead of re-reading the same prefix forever. +func TestClient_GetAllPendingOutbounds_CarriesCursorAcrossPolls(t *testing.T) { + ctx := context.Background() + + total := pendingOutboundMaxPages + 3 + pages := make([]*uexecutortypes.QueryAllPendingOutboundsResponse, total) + for i := range pages { + var next []byte + if i < total-1 { + next = []byte(fmt.Sprintf("k%d", i+1)) + } + pages[i] = pendingPage(fmt.Sprintf("ob-%d", i), next) + } + mockClient := &mockUExecutorQueryClient{pages: pages} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + // First poll spends the budget and parks mid-set. + first, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + require.Len(t, first, pendingOutboundMaxPages) + require.NotNil(t, client.pendingCursor, "must remember where it stopped") + + // Second poll resumes from there rather than restarting at the head. + resumeFrom := client.pendingCursor + second, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + require.Len(t, second, 3, "the remainder of the set") + assert.Equal(t, "ob-"+fmt.Sprint(pendingOutboundMaxPages), second[0].OutboundId, + "must continue after the parked cursor, not re-read the prefix") + assert.Equal(t, resumeFrom, mockClient.requestedKeys[pendingOutboundMaxPages], + "the parked cursor is what gets sent") + + // Reaching the end resets, so tail additions are seen on the next poll. + assert.Nil(t, client.pendingCursor) +} + +// A failed page must not lose the ground already covered either. +func TestClient_GetAllPendingOutbounds_ResumesAfterPageFailure(t *testing.T) { + mockClient := &mockUExecutorQueryClient{ + pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ + pendingPage("ob-0", []byte("k1")), + pendingPage("ob-1", []byte("k2")), + }, + failAfterPage: 1, + } + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + entries, _, err := client.GetAllPendingOutbounds(context.Background()) + require.NoError(t, err) + require.Len(t, entries, 1) + assert.Equal(t, []byte("k1"), client.pendingCursor, "resume at the page that failed") +} From 796dd126817a6782808987fc5e9374fd10bbc94a Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 17:33:52 +0530 Subject: [PATCH 4/8] fix: request the full pending outbound set; the server pages by offset and returns no next key (F-2026-18817) --- universalClient/pushcore/pushCore.go | 103 ++++----- universalClient/pushcore/pushCore_test.go | 249 +++------------------- 2 files changed, 71 insertions(+), 281 deletions(-) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index d82280b5..1302217e 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -9,7 +9,6 @@ import ( "fmt" "math/big" "strings" - "sync" "sync/atomic" cmtservice "github.com/cosmos/cosmos-sdk/client/grpc/cmtservice" @@ -41,13 +40,6 @@ type Client struct { authClients []authtypes.QueryClient // Auth query clients conns []*grpc.ClientConn // Owned gRPC connections (for cleanup) rr uint32 // Round-robin counter for endpoint selection - - // Pagination cursor carried between pending-outbound polls. The page budget - // bounds one poll's work; this makes the budget cost latency rather than - // coverage, so a set larger than the budget is still walked in full over - // successive ticks. Nil means start from the beginning. - pendingMu sync.Mutex - pendingCursor []byte } // New creates a new Client by dialing the provided gRPC URLs. @@ -405,8 +397,14 @@ func (c *Client) GetPendingFundMigrations(ctx context.Context) ([]*utsstypes.Fun // 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 ( - pendingOutboundPageSize = 1000 - pendingOutboundMaxPages = 20 + // 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. + pendingOutboundLimit = 100_000 chainConfigPageSize = 200 chainConfigMaxPages = 20 @@ -415,68 +413,39 @@ const ( // GetAllPendingOutbounds retrieves pending outbound transactions from Push Chain, // sorted by created_at (block height) ascending — oldest first. // -// The result is paged rather than a single query. An outbound only leaves the -// pending set once a quorum vote terminalizes it, so any row that cannot reach -// one — for example a destination execution whose observation was never seen — -// stays at the head of an oldest-first list forever. Reading one fixed page -// would let such a prefix hide every newer outbound from signing, on every -// chain, since this query is not chain-scoped. +// 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. +// The request must therefore cover the whole set: a fixed small page would let +// such a prefix hide every newer outbound from signing, on every chain, since +// this query is not chain-scoped. func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes.PendingOutboundEntry, []*uexecutortypes.OutboundTx, error) { - c.pendingMu.Lock() - defer c.pendingMu.Unlock() - - var ( - entries []*uexecutortypes.PendingOutboundEntry - outbounds []*uexecutortypes.OutboundTx - nextKey = c.pendingCursor + 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: pendingOutboundLimit}, + }) + }, + "GetAllPendingOutbounds", + c.logger, ) + if err != nil { + return nil, nil, err + } - for page := 0; page < pendingOutboundMaxPages; page++ { - key := nextKey - 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{Key: key, Limit: pendingOutboundPageSize}, - }) - }, - "GetAllPendingOutbounds", - c.logger, - ) - if err != nil { - // Return what we have rather than nothing: a later page failing must - // not stop the caller acting on the pages that did arrive. Resume from - // the failed page next time instead of losing the ground already made. - if len(entries) > 0 { - c.pendingCursor = key - c.logger.Warn().Err(err).Int("page", page).Msg("pending outbound page failed, using pages fetched so far") - return entries, outbounds, nil - } - c.pendingCursor = nil - return nil, nil, err - } - - entries = append(entries, resp.Entries...) - outbounds = append(outbounds, resp.Outbounds...) - - if resp.Pagination == nil || len(resp.Pagination.NextKey) == 0 { - // Reached the end; the next poll starts from the beginning again so - // rows added at the tail since the walk began are picked up. - c.pendingCursor = nil - return entries, outbounds, nil - } - nextKey = resp.Pagination.NextKey + // Total is authoritative for the size of the pending set, so a shortfall means + // rows we will not act on this poll. Loud rather than silent: those outbounds + // are invisible to signing until the set shrinks. + if resp.Pagination != nil && resp.Pagination.Total > uint64(len(resp.Entries)) { + c.logger.Error(). + Uint64("total", resp.Pagination.Total). + Int("received", len(resp.Entries)). + Uint64("limit", pendingOutboundLimit). + Msg("pending outbound set exceeds the request limit; the remainder is not being signed") } - // Budget spent mid-set. Park the cursor so the next tick continues from here - // rather than re-reading the same prefix forever. - c.pendingCursor = nextKey - c.logger.Warn(). - Int("max_pages", pendingOutboundMaxPages). - Int("fetched", len(entries)). - Msg("pending outbound page budget spent; continuing from this cursor next poll") - return entries, outbounds, nil + return resp.Entries, resp.Outbounds, nil } // createGRPCConnection creates a gRPC connection with appropriate transport security. diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index 9fcb1439..1e149205 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -1,6 +1,7 @@ package pushcore import ( + "bytes" "context" "errors" "fmt" @@ -1046,11 +1047,8 @@ type mockUExecutorQueryClient struct { allPendingOutboundsResp *uexecutortypes.QueryAllPendingOutboundsResponse err error - // Paging support: `pages` is served in order, one per call, and the key the - // caller sent is recorded so tests can assert NextKey is actually followed. - pages []*uexecutortypes.QueryAllPendingOutboundsResponse - requestedKeys [][]byte - failAfterPage int // when > 0, calls beyond this page number return err + // 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) { @@ -1077,21 +1075,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) { - if m.pages != nil { - var key []byte - if req.Pagination != nil { - key = req.Pagination.Key - } - m.requestedKeys = append(m.requestedKeys, key) - idx := len(m.requestedKeys) - 1 - if m.failAfterPage > 0 && idx >= m.failAfterPage { - return nil, assert.AnError - } - if idx >= len(m.pages) { - return nil, assert.AnError - } - return m.pages[idx], nil - } + m.lastPendingReq = req if m.err != nil { return nil, m.err } @@ -1188,217 +1172,54 @@ func TestClient_GetKeyByID(t *testing.T) { }) } -// page builds one response carrying a single outbound, with nextKey signalling -// whether more pages follow. -func pendingPage(id string, nextKey []byte) *uexecutortypes.QueryAllPendingOutboundsResponse { - return &uexecutortypes.QueryAllPendingOutboundsResponse{ - Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: id, UniversalTxId: "utx-" + id}}, - Outbounds: []*uexecutortypes.OutboundTx{{Id: id, DestinationChain: "eip155:1"}}, - Pagination: &query.PageResponse{NextKey: nextKey}, - } -} - -// Pending outbounds are oldest-first and only leave the set on a quorum vote, so -// a row that can never reach one sits at the head permanently. Reading a single -// page would hide every newer outbound behind it, on every chain. -func TestClient_GetAllPendingOutbounds_Paginates(t *testing.T) { +// This server pages by offset and returns only Total, never a NextKey, and its +// sort is unstable on a block height so offset boundaries can shift between +// calls. One request covering the set, plus a Total check, is the only shape +// that is correct against it. +func TestClient_GetAllPendingOutbounds_CoversTheWholeSet(t *testing.T) { ctx := context.Background() - t.Run("follows NextKey until exhausted", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ - pendingPage("stuck", []byte("k1")), - pendingPage("also-stuck", []byte("k2")), - pendingPage("reachable", nil), // last page - }} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - entries, outbounds, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - require.Len(t, entries, 3) - require.Len(t, outbounds, 3) - - // The outbound behind the stuck prefix must be visible. - assert.Equal(t, "reachable", entries[2].OutboundId) - - // And the cursor must actually be threaded through, not just re-queried. - assert.Equal(t, [][]byte{nil, []byte("k1"), []byte("k2")}, mockClient.requestedKeys) - }) + resp := func(n int, total uint64) *uexecutortypes.QueryAllPendingOutboundsResponse { + r := &uexecutortypes.QueryAllPendingOutboundsResponse{Pagination: &query.PageResponse{Total: total}} + for i := 0; i < n; i++ { + r.Entries = append(r.Entries, &uexecutortypes.PendingOutboundEntry{OutboundId: fmt.Sprintf("ob-%d", i)}) + r.Outbounds = append(r.Outbounds, &uexecutortypes.OutboundTx{Id: fmt.Sprintf("ob-%d", i)}) + } + return r + } - t.Run("single page with no NextKey stops immediately", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ - pendingPage("only", nil), - }} + t.Run("requests a limit covering the set", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(1, 1)} client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} entries, _, err := client.GetAllPendingOutbounds(ctx) require.NoError(t, err) require.Len(t, entries, 1) - assert.Len(t, mockClient.requestedKeys, 1, "must not request a second page") + require.NotNil(t, mockClient.lastPendingReq.Pagination) + assert.Equal(t, uint64(pendingOutboundLimit), mockClient.lastPendingReq.Pagination.Limit, + "a small page would let a stuck prefix hide newer outbounds") }) - t.Run("nil pagination stops the walk", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ - {Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: "ob-1"}}, - Outbounds: []*uexecutortypes.OutboundTx{{Id: "ob-1"}}}, - }} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + // A shortfall means outbounds nobody is signing, so it has to be visible. + t.Run("reports a set larger than the limit", func(t *testing.T) { + var logBuf bytes.Buffer + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(1, 5)} + client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} entries, _, err := client.GetAllPendingOutbounds(ctx) require.NoError(t, err) require.Len(t, entries, 1) - assert.Len(t, mockClient.requestedKeys, 1) - }) - - // A later page failing must not discard the earlier ones: those outbounds are - // still signable, and returning nothing would stall them for no reason. - t.Run("mid-walk failure keeps the pages already fetched", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{ - pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ - pendingPage("ob-1", []byte("k1")), - pendingPage("ob-2", []byte("k2")), - }, - failAfterPage: 1, - } - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - entries, outbounds, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - require.Len(t, entries, 1) - require.Len(t, outbounds, 1) - assert.Equal(t, "ob-1", entries[0].OutboundId) - }) - - t.Run("first page failing is an error", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{ - pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{pendingPage("ob-1", nil)}, - failAfterPage: 0, - err: assert.AnError, - } - mockClient.pages = nil // fall through to the plain error path - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - entries, _, err := client.GetAllPendingOutbounds(ctx) - require.Error(t, err) - assert.Nil(t, entries) - }) - - // The cap bounds one poll; the rest is picked up next tick rather than - // growing the request without limit. - t.Run("stops at the page cap", func(t *testing.T) { - pages := make([]*uexecutortypes.QueryAllPendingOutboundsResponse, pendingOutboundMaxPages+5) - for i := range pages { - pages[i] = pendingPage(fmt.Sprintf("ob-%d", i), []byte("more")) - } - mockClient := &mockUExecutorQueryClient{pages: pages} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - entries, _, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - assert.Len(t, entries, pendingOutboundMaxPages) - assert.Len(t, mockClient.requestedKeys, pendingOutboundMaxPages) - }) -} - -func chainConfigPage(chain string, nextKey []byte) *uregistrytypes.QueryAllChainConfigsResponse { - return &uregistrytypes.QueryAllChainConfigsResponse{ - Configs: []*uregistrytypes.ChainConfig{{Chain: chain}}, - Pagination: &query.PageResponse{NextKey: nextKey}, - } -} - -// The server paginates this collection, so omitting a PageRequest capped the -// response at the SDK default of 100. A chain missing from the list is simply -// never watched, so the walk must not be able to truncate. -func TestClient_GetAllChainConfigs_Paginates(t *testing.T) { - ctx := context.Background() - - t.Run("follows NextKey across pages", func(t *testing.T) { - mockClient := &mockRegistryQueryClient{chainConfigPages: []*uregistrytypes.QueryAllChainConfigsResponse{ - chainConfigPage("eip155:1", []byte("k1")), - chainConfigPage("solana:x", nil), - }} - client := &Client{logger: zerolog.Nop(), eps: []uregistrytypes.QueryClient{mockClient}} - - configs, err := client.GetAllChainConfigs(ctx) - require.NoError(t, err) - require.Len(t, configs, 2) - assert.Equal(t, "solana:x", configs[1].Chain) - assert.Equal(t, [][]byte{nil, []byte("k1")}, mockClient.chainConfigKeys) + assert.Contains(t, logBuf.String(), "exceeds the request limit") + assert.Contains(t, logBuf.String(), `"total":5`) }) - t.Run("a page request is always sent", func(t *testing.T) { - mockClient := &mockRegistryQueryClient{chainConfigPages: []*uregistrytypes.QueryAllChainConfigsResponse{ - chainConfigPage("eip155:1", nil), - }} - client := &Client{logger: zerolog.Nop(), eps: []uregistrytypes.QueryClient{mockClient}} + t.Run("quiet when the set is fully covered", func(t *testing.T) { + var logBuf bytes.Buffer + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(2, 2)} + client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - _, err := client.GetAllChainConfigs(ctx) + _, _, err := client.GetAllPendingOutbounds(ctx) require.NoError(t, err) - require.Len(t, mockClient.chainConfigKeys, 1, "must not request a second page") - }) - - t.Run("error propagates", func(t *testing.T) { - client := &Client{logger: zerolog.Nop(), eps: []uregistrytypes.QueryClient{ - &mockRegistryQueryClient{err: assert.AnError}, - }} - - configs, err := client.GetAllChainConfigs(ctx) - require.Error(t, err) - assert.Nil(t, configs) + assert.NotContains(t, logBuf.String(), "exceeds the request limit") }) } - -// The page budget bounds one poll's work, so it must cost latency rather than -// coverage: a set larger than the budget has to be walked in full across -// successive polls instead of re-reading the same prefix forever. -func TestClient_GetAllPendingOutbounds_CarriesCursorAcrossPolls(t *testing.T) { - ctx := context.Background() - - total := pendingOutboundMaxPages + 3 - pages := make([]*uexecutortypes.QueryAllPendingOutboundsResponse, total) - for i := range pages { - var next []byte - if i < total-1 { - next = []byte(fmt.Sprintf("k%d", i+1)) - } - pages[i] = pendingPage(fmt.Sprintf("ob-%d", i), next) - } - mockClient := &mockUExecutorQueryClient{pages: pages} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - // First poll spends the budget and parks mid-set. - first, _, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - require.Len(t, first, pendingOutboundMaxPages) - require.NotNil(t, client.pendingCursor, "must remember where it stopped") - - // Second poll resumes from there rather than restarting at the head. - resumeFrom := client.pendingCursor - second, _, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - require.Len(t, second, 3, "the remainder of the set") - assert.Equal(t, "ob-"+fmt.Sprint(pendingOutboundMaxPages), second[0].OutboundId, - "must continue after the parked cursor, not re-read the prefix") - assert.Equal(t, resumeFrom, mockClient.requestedKeys[pendingOutboundMaxPages], - "the parked cursor is what gets sent") - - // Reaching the end resets, so tail additions are seen on the next poll. - assert.Nil(t, client.pendingCursor) -} - -// A failed page must not lose the ground already covered either. -func TestClient_GetAllPendingOutbounds_ResumesAfterPageFailure(t *testing.T) { - mockClient := &mockUExecutorQueryClient{ - pages: []*uexecutortypes.QueryAllPendingOutboundsResponse{ - pendingPage("ob-0", []byte("k1")), - pendingPage("ob-1", []byte("k2")), - }, - failAfterPage: 1, - } - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - entries, _, err := client.GetAllPendingOutbounds(context.Background()) - require.NoError(t, err) - require.Len(t, entries, 1) - assert.Equal(t, []byte("k1"), client.pendingCursor, "resume at the page that failed") -} From e82d7bdc62e382e09d6d338bb5648f925aaacd58 Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 17:41:57 +0530 Subject: [PATCH 5/8] fix: raise grpc receive limit so a large pending set does not fail the whole poll (F-2026-18817) --- universalClient/pushcore/pushCore.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index 1302217e..a022716f 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -406,6 +406,12 @@ const ( // that is both correct and cheap against that server. pendingOutboundLimit = 100_000 + // gRPC receive cap. The default is 4 MiB, which the pending-outbound response + // outgrows at a few thousand rows; exceeding it fails the call rather than + // truncating, so the poll stops entirely. 64 MiB keeps the transport from + // being the binding constraint on a set the server will happily return. + maxCallRecvMsgSize = 64 * 1024 * 1024 + chainConfigPageSize = 200 chainConfigMaxPages = 20 ) @@ -485,6 +491,14 @@ func createGRPCConnection(endpoint string) (*grpc.ClientConn, error) { opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) } + // The pending-outbound query returns the whole set in one response and a row + // costs on the order of a kilobyte, so gRPC's 4 MiB default caps it at a few + // thousand rows — and it caps by failing the call outright, which takes the + // entire poll down rather than returning a short list we could detect. + opts = append(opts, grpc.WithDefaultCallOptions( + grpc.MaxCallRecvMsgSize(maxCallRecvMsgSize), + )) + conn, err := grpc.NewClient(processedEndpoint, opts...) if err != nil { return nil, fmt.Errorf("failed to create gRPC connection to %s: %w", processedEndpoint, err) From 332b09f6b73a8cbf77a613b91a35d31e9c792174 Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 17:44:39 +0530 Subject: [PATCH 6/8] fix: alternate pending outbound sweep direction instead of raising transport limits (F-2026-18817) --- universalClient/pushcore/pushCore.go | 32 +++++++++++------------ universalClient/pushcore/pushCore_test.go | 27 +++++++++++++++++-- 2 files changed, 40 insertions(+), 19 deletions(-) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index a022716f..7ed275d6 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -40,6 +40,7 @@ type Client struct { authClients []authtypes.QueryClient // Auth query clients conns []*grpc.ClientConn // Owned gRPC connections (for cleanup) rr uint32 // Round-robin counter for endpoint selection + pendingSweep uint32 // Alternates pending-outbound sort direction per poll } // New creates a new Client by dialing the provided gRPC URLs. @@ -404,13 +405,10 @@ const ( // 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. - pendingOutboundLimit = 100_000 - - // gRPC receive cap. The default is 4 MiB, which the pending-outbound response - // outgrows at a few thousand rows; exceeding it fails the call rather than - // truncating, so the poll stops entirely. 64 MiB keeps the transport from - // being the binding constraint on a set the server will happily return. - maxCallRecvMsgSize = 64 * 1024 * 1024 + // 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 @@ -425,12 +423,19 @@ const ( // such a prefix hide every newer outbound from signing, on every chain, since // this query is not chain-scoped. func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes.PendingOutboundEntry, []*uexecutortypes.OutboundTx, error) { + // Alternate direction per poll. New outbounds always arrive at the newest end, + // so a newest-first read stays useful however many unterminalized rows have + // piled up at the oldest end; the oldest-first read still covers the backlog + // and a cold local database. Rows seen in both directions are deduplicated by + // the caller, so the overlap costs nothing. + reverse := atomic.AddUint32(&c.pendingSweep, 1)%2 == 1 + 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: pendingOutboundLimit}, + Pagination: &query.PageRequest{Limit: pendingOutboundLimit, Reverse: reverse}, }) }, "GetAllPendingOutbounds", @@ -448,7 +453,8 @@ func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes. Uint64("total", resp.Pagination.Total). Int("received", len(resp.Entries)). Uint64("limit", pendingOutboundLimit). - Msg("pending outbound set exceeds the request limit; the remainder is not being signed") + Bool("newest_first", reverse). + Msg("pending outbound set exceeds the request limit; the far end is only seen on the alternate sweep") } return resp.Entries, resp.Outbounds, nil @@ -491,14 +497,6 @@ func createGRPCConnection(endpoint string) (*grpc.ClientConn, error) { opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) } - // The pending-outbound query returns the whole set in one response and a row - // costs on the order of a kilobyte, so gRPC's 4 MiB default caps it at a few - // thousand rows — and it caps by failing the call outright, which takes the - // entire poll down rather than returning a short list we could detect. - opts = append(opts, grpc.WithDefaultCallOptions( - grpc.MaxCallRecvMsgSize(maxCallRecvMsgSize), - )) - conn, err := grpc.NewClient(processedEndpoint, opts...) if err != nil { return nil, fmt.Errorf("failed to create gRPC connection to %s: %w", processedEndpoint, err) diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index 1e149205..a26274f4 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -1196,8 +1196,7 @@ func TestClient_GetAllPendingOutbounds_CoversTheWholeSet(t *testing.T) { require.NoError(t, err) require.Len(t, entries, 1) require.NotNil(t, mockClient.lastPendingReq.Pagination) - assert.Equal(t, uint64(pendingOutboundLimit), mockClient.lastPendingReq.Pagination.Limit, - "a small page would let a stuck prefix hide newer outbounds") + assert.Equal(t, uint64(pendingOutboundLimit), mockClient.lastPendingReq.Pagination.Limit) }) // A shortfall means outbounds nobody is signing, so it has to be visible. @@ -1223,3 +1222,27 @@ func TestClient_GetAllPendingOutbounds_CoversTheWholeSet(t *testing.T) { assert.NotContains(t, logBuf.String(), "exceeds the request limit") }) } + +// A stuck prefix only ever builds at the oldest end, and new outbounds only ever +// arrive at the newest end. Alternating the sweep means neither end can be +// hidden by the other, without asking for a set large enough to break the +// transport. +func TestClient_GetAllPendingOutbounds_AlternatesSweepDirection(t *testing.T) { + mockClient := &mockUExecutorQueryClient{ + allPendingOutboundsResp: &uexecutortypes.QueryAllPendingOutboundsResponse{ + Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: "ob-1"}}, + Outbounds: []*uexecutortypes.OutboundTx{{Id: "ob-1"}}, + Pagination: &query.PageResponse{Total: 1}, + }, + } + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + seen := make([]bool, 0, 4) + for i := 0; i < 4; i++ { + _, _, err := client.GetAllPendingOutbounds(context.Background()) + require.NoError(t, err) + seen = append(seen, mockClient.lastPendingReq.Pagination.Reverse) + } + + assert.Equal(t, []bool{true, false, true, false}, seen, "each poll must flip the direction") +} From 061d476179352c6e3e06e5e422d2312eed34ad3a Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 21:46:20 +0530 Subject: [PATCH 7/8] fix: read newest-first every poll, sweep the backlog only when it exceeds one request (F-2026-18817) --- universalClient/pushcore/pushCore.go | 78 ++++++++++++--- universalClient/pushcore/pushCore_test.go | 114 +++++++++++++++++----- 2 files changed, 151 insertions(+), 41 deletions(-) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index 7ed275d6..f1df15cb 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -9,6 +9,7 @@ import ( "fmt" "math/big" "strings" + "sync" "sync/atomic" cmtservice "github.com/cosmos/cosmos-sdk/client/grpc/cmtservice" @@ -40,7 +41,13 @@ type Client struct { authClients []authtypes.QueryClient // Auth query clients conns []*grpc.ClientConn // Owned gRPC connections (for cleanup) rr uint32 // Round-robin counter for endpoint selection - pendingSweep uint32 // Alternates pending-outbound sort direction per poll + + // Pending-outbound read state. Only consulted when the set outgrows a single + // request; below that the newest-first read is already complete. + pendingMu sync.Mutex + pendingPoll uint64 // poll counter, alternates newest-first with a backlog sweep + pendingTotal uint64 // set size reported by the last poll + backlogOffset uint64 // rotating offset for the oldest-first sweep } // New creates a new Client by dialing the provided gRPC URLs. @@ -423,19 +430,14 @@ const ( // such a prefix hide every newer outbound from signing, on every chain, since // this query is not chain-scoped. func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes.PendingOutboundEntry, []*uexecutortypes.OutboundTx, error) { - // Alternate direction per poll. New outbounds always arrive at the newest end, - // so a newest-first read stays useful however many unterminalized rows have - // piled up at the oldest end; the oldest-first read still covers the backlog - // and a cold local database. Rows seen in both directions are deduplicated by - // the caller, so the overlap costs nothing. - reverse := atomic.AddUint32(&c.pendingSweep, 1)%2 == 1 + reverse, offset := c.nextPendingRead() 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: pendingOutboundLimit, Reverse: reverse}, + Pagination: &query.PageRequest{Limit: pendingOutboundLimit, Reverse: reverse, Offset: offset}, }) }, "GetAllPendingOutbounds", @@ -445,21 +447,65 @@ func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes. return nil, nil, err } - // Total is authoritative for the size of the pending set, so a shortfall means - // rows we will not act on this poll. Loud rather than silent: those outbounds - // are invisible to signing until the set shrinks. - if resp.Pagination != nil && resp.Pagination.Total > uint64(len(resp.Entries)) { - c.logger.Error(). - Uint64("total", resp.Pagination.Total). - Int("received", len(resp.Entries)). + var total uint64 + if resp.Pagination != nil { + total = resp.Pagination.Total + } + c.recordPendingTotal(total) + + if total > pendingOutboundLimit { + c.logger.Warn(). + Uint64("total", total). Uint64("limit", pendingOutboundLimit). Bool("newest_first", reverse). - Msg("pending outbound set exceeds the request limit; the far end is only seen on the alternate sweep") + Uint64("offset", offset). + Msg("pending outbound set exceeds one request; newest rows are read every poll, the backlog is swept in slices") } return resp.Entries, resp.Outbounds, nil } +// nextPendingRead picks the direction and offset for this poll. +// +// New outbounds only ever arrive at the newest end, so a newest-first read at +// offset zero always surfaces them however many unterminalized rows have piled +// up at the oldest end. That read alone is complete whenever the set fits in one +// request, which is the normal case. +// +// Once it does not fit, alternate polls sweep the backlog oldest-first at an +// advancing offset. That exists only so a node that started with a large pending +// set — an empty local database, for instance — eventually sees the older rows; +// steady-state discovery does not need it. Offsets over a set that is being +// added to and terminalized can occasionally straddle a row, which is why the +// sweep keeps cycling rather than running once. +func (c *Client) nextPendingRead() (reverse bool, offset uint64) { + c.pendingMu.Lock() + defer c.pendingMu.Unlock() + + c.pendingPoll++ + if c.pendingTotal <= pendingOutboundLimit || c.pendingPoll%2 == 1 { + return true, 0 + } + + offset = c.backlogOffset + c.backlogOffset += pendingOutboundLimit + if c.backlogOffset >= c.pendingTotal { + c.backlogOffset = 0 + } + return false, offset +} + +// recordPendingTotal keeps the observed set size so the next poll can tell +// whether a backlog sweep is needed at all. +func (c *Client) recordPendingTotal(total uint64) { + c.pendingMu.Lock() + defer c.pendingMu.Unlock() + c.pendingTotal = total + if total <= pendingOutboundLimit { + c.backlogOffset = 0 + } +} + // createGRPCConnection creates a gRPC connection with appropriate transport security. // It automatically detects whether to use TLS based on the URL scheme // and adds default port 9090 if no port is specified. diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index a26274f4..ab338f98 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -1199,17 +1199,16 @@ func TestClient_GetAllPendingOutbounds_CoversTheWholeSet(t *testing.T) { assert.Equal(t, uint64(pendingOutboundLimit), mockClient.lastPendingReq.Pagination.Limit) }) - // A shortfall means outbounds nobody is signing, so it has to be visible. - t.Run("reports a set larger than the limit", func(t *testing.T) { + // A set too large for one request means the backlog is only seen in slices, + // which an operator should know about. + t.Run("reports a set larger than one request", func(t *testing.T) { var logBuf bytes.Buffer - mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(1, 5)} + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(1, pendingOutboundLimit+2500)} client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - entries, _, err := client.GetAllPendingOutbounds(ctx) + _, _, err := client.GetAllPendingOutbounds(ctx) require.NoError(t, err) - require.Len(t, entries, 1) - assert.Contains(t, logBuf.String(), "exceeds the request limit") - assert.Contains(t, logBuf.String(), `"total":5`) + assert.Contains(t, logBuf.String(), "exceeds one request") }) t.Run("quiet when the set is fully covered", func(t *testing.T) { @@ -1223,26 +1222,91 @@ func TestClient_GetAllPendingOutbounds_CoversTheWholeSet(t *testing.T) { }) } -// A stuck prefix only ever builds at the oldest end, and new outbounds only ever -// arrive at the newest end. Alternating the sweep means neither end can be -// hidden by the other, without asking for a set large enough to break the -// transport. -func TestClient_GetAllPendingOutbounds_AlternatesSweepDirection(t *testing.T) { - mockClient := &mockUExecutorQueryClient{ - allPendingOutboundsResp: &uexecutortypes.QueryAllPendingOutboundsResponse{ - Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: "ob-1"}}, - Outbounds: []*uexecutortypes.OutboundTx{{Id: "ob-1"}}, - Pagination: &query.PageResponse{Total: 1}, - }, +// pendingResp builds a response reporting `total` as the size of the pending set. +func pendingResp(total uint64) *uexecutortypes.QueryAllPendingOutboundsResponse { + return &uexecutortypes.QueryAllPendingOutboundsResponse{ + Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: "ob-1"}}, + Outbounds: []*uexecutortypes.OutboundTx{{Id: "ob-1"}}, + Pagination: &query.PageResponse{Total: total}, } - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} +} - seen := make([]bool, 0, 4) - for i := 0; i < 4; i++ { - _, _, err := client.GetAllPendingOutbounds(context.Background()) - require.NoError(t, err) - seen = append(seen, mockClient.lastPendingReq.Pagination.Reverse) +// New outbounds only arrive at the newest end, so that read is what must never +// be crowded out. The backlog sweep exists only for a set too large to fetch at +// once, and reading it at a fixed offset would re-return the same prefix forever. +func TestClient_GetAllPendingOutbounds_ReadStrategy(t *testing.T) { + ctx := context.Background() + + read := func(m *mockUExecutorQueryClient) (bool, uint64) { + p := m.lastPendingReq.Pagination + return p.Reverse, p.Offset } - assert.Equal(t, []bool{true, false, true, false}, seen, "each poll must flip the direction") + t.Run("set fits in one request: always newest-first, never sweeps", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(9)} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + for i := 0; i < 4; i++ { + _, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + rev, off := read(mockClient) + assert.True(t, rev, "poll %d should read newest-first", i) + assert.Zero(t, off, "poll %d should not sweep a set that fits", i) + } + }) + + t.Run("set too large: newest-first every other poll, backlog advances", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(3500)} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + var got [][2]any + for i := 0; i < 6; i++ { + _, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + rev, off := read(mockClient) + got = append(got, [2]any{rev, off}) + } + + // poll 1 has no prior total, so it starts newest-first and the sweep + // engages once the size is known. + assert.Equal(t, [2]any{true, uint64(0)}, got[0]) + assert.Equal(t, [2]any{false, uint64(0)}, got[1], "backlog sweep starts at the oldest end") + assert.Equal(t, [2]any{true, uint64(0)}, got[2], "newest rows read every other poll") + assert.Equal(t, [2]any{false, uint64(1000)}, got[3], "sweep advances") + assert.Equal(t, [2]any{true, uint64(0)}, got[4]) + assert.Equal(t, [2]any{false, uint64(2000)}, got[5], "and keeps advancing") + }) + + t.Run("sweep wraps rather than running off the end", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(2500)} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + var offsets []uint64 + for i := 0; i < 10; i++ { + _, _, err := client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + if rev, off := read(mockClient); !rev { + offsets = append(offsets, off) + } + } + assert.Equal(t, []uint64{0, 1000, 2000, 0, 1000}, offsets, "covers the set then starts over") + }) + + t.Run("sweep stops once the set shrinks to fit", func(t *testing.T) { + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(3500)} + client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} + + _, _, err := client.GetAllPendingOutbounds(ctx) // learns the size + require.NoError(t, err) + mockClient.allPendingOutboundsResp = pendingResp(12) // backlog cleared + + _, _, err = client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + _, _, err = client.GetAllPendingOutbounds(ctx) + require.NoError(t, err) + + rev, off := read(mockClient) + assert.True(t, rev) + assert.Zero(t, off, "no sweep needed once one request covers the set") + }) } From 51219137447605d154facdaf863ad70f8d133dc8 Mon Sep 17 00:00:00 2001 From: aman035 Date: Thu, 20 Aug 2026 22:24:32 +0530 Subject: [PATCH 8/8] simplify: read pending outbounds newest-first, drop the backlog sweep machinery (F-2026-18817) --- universalClient/pushcore/pushCore.go | 89 +++----------- universalClient/pushcore/pushCore_test.go | 138 ++++------------------ 2 files changed, 44 insertions(+), 183 deletions(-) diff --git a/universalClient/pushcore/pushCore.go b/universalClient/pushcore/pushCore.go index f1df15cb..4a2393c8 100644 --- a/universalClient/pushcore/pushCore.go +++ b/universalClient/pushcore/pushCore.go @@ -9,7 +9,6 @@ import ( "fmt" "math/big" "strings" - "sync" "sync/atomic" cmtservice "github.com/cosmos/cosmos-sdk/client/grpc/cmtservice" @@ -41,13 +40,6 @@ type Client struct { authClients []authtypes.QueryClient // Auth query clients conns []*grpc.ClientConn // Owned gRPC connections (for cleanup) rr uint32 // Round-robin counter for endpoint selection - - // Pending-outbound read state. Only consulted when the set outgrows a single - // request; below that the newest-first read is already complete. - pendingMu sync.Mutex - pendingPoll uint64 // poll counter, alternates newest-first with a backlog sweep - pendingTotal uint64 // set size reported by the last poll - backlogOffset uint64 // rotating offset for the oldest-first sweep } // New creates a new Client by dialing the provided gRPC URLs. @@ -421,23 +413,25 @@ const ( chainConfigMaxPages = 20 ) -// GetAllPendingOutbounds retrieves pending outbound transactions from Push Chain, -// sorted by created_at (block height) ascending — oldest first. +// 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. // -// 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. -// The request must therefore cover the whole set: a fixed small page would let -// such a prefix hide every newer outbound from signing, on every chain, since -// this query is not chain-scoped. +// 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) { - reverse, offset := c.nextPendingRead() - 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: pendingOutboundLimit, Reverse: reverse, Offset: offset}, + Pagination: &query.PageRequest{Limit: pendingOutboundLimit, Reverse: true}, }) }, "GetAllPendingOutbounds", @@ -447,65 +441,20 @@ func (c *Client) GetAllPendingOutbounds(ctx context.Context) ([]*uexecutortypes. return nil, nil, err } - var total uint64 - if resp.Pagination != nil { - total = resp.Pagination.Total - } - c.recordPendingTotal(total) - - if total > pendingOutboundLimit { + // 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", total). + Uint64("total", resp.Pagination.Total). Uint64("limit", pendingOutboundLimit). - Bool("newest_first", reverse). - Uint64("offset", offset). - Msg("pending outbound set exceeds one request; newest rows are read every poll, the backlog is swept in slices") + 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 } -// nextPendingRead picks the direction and offset for this poll. -// -// New outbounds only ever arrive at the newest end, so a newest-first read at -// offset zero always surfaces them however many unterminalized rows have piled -// up at the oldest end. That read alone is complete whenever the set fits in one -// request, which is the normal case. -// -// Once it does not fit, alternate polls sweep the backlog oldest-first at an -// advancing offset. That exists only so a node that started with a large pending -// set — an empty local database, for instance — eventually sees the older rows; -// steady-state discovery does not need it. Offsets over a set that is being -// added to and terminalized can occasionally straddle a row, which is why the -// sweep keeps cycling rather than running once. -func (c *Client) nextPendingRead() (reverse bool, offset uint64) { - c.pendingMu.Lock() - defer c.pendingMu.Unlock() - - c.pendingPoll++ - if c.pendingTotal <= pendingOutboundLimit || c.pendingPoll%2 == 1 { - return true, 0 - } - - offset = c.backlogOffset - c.backlogOffset += pendingOutboundLimit - if c.backlogOffset >= c.pendingTotal { - c.backlogOffset = 0 - } - return false, offset -} - -// recordPendingTotal keeps the observed set size so the next poll can tell -// whether a backlog sweep is needed at all. -func (c *Client) recordPendingTotal(total uint64) { - c.pendingMu.Lock() - defer c.pendingMu.Unlock() - c.pendingTotal = total - if total <= pendingOutboundLimit { - c.backlogOffset = 0 - } -} - // createGRPCConnection creates a gRPC connection with appropriate transport security. // It automatically detects whether to use TLS based on the URL scheme // and adds default port 9090 if no port is specified. diff --git a/universalClient/pushcore/pushCore_test.go b/universalClient/pushcore/pushCore_test.go index ab338f98..cbcd6fb4 100644 --- a/universalClient/pushcore/pushCore_test.go +++ b/universalClient/pushcore/pushCore_test.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "errors" - "fmt" "math/big" "testing" @@ -1172,38 +1171,40 @@ func TestClient_GetKeyByID(t *testing.T) { }) } -// This server pages by offset and returns only Total, never a NextKey, and its -// sort is unstable on a block height so offset boundaries can shift between -// calls. One request covering the set, plus a Total check, is the only shape -// that is correct against it. -func TestClient_GetAllPendingOutbounds_CoversTheWholeSet(t *testing.T) { +// 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(n int, total uint64) *uexecutortypes.QueryAllPendingOutboundsResponse { - r := &uexecutortypes.QueryAllPendingOutboundsResponse{Pagination: &query.PageResponse{Total: total}} - for i := 0; i < n; i++ { - r.Entries = append(r.Entries, &uexecutortypes.PendingOutboundEntry{OutboundId: fmt.Sprintf("ob-%d", i)}) - r.Outbounds = append(r.Outbounds, &uexecutortypes.OutboundTx{Id: fmt.Sprintf("ob-%d", i)}) + 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}, } - return r } - t.Run("requests a limit covering the set", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(1, 1)} + 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}} - entries, _, err := client.GetAllPendingOutbounds(ctx) + _, _, err := client.GetAllPendingOutbounds(ctx) require.NoError(t, err) - require.Len(t, entries, 1) - require.NotNil(t, mockClient.lastPendingReq.Pagination) - assert.Equal(t, uint64(pendingOutboundLimit), mockClient.lastPendingReq.Pagination.Limit) + + 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) }) - // A set too large for one request means the backlog is only seen in slices, - // which an operator should know about. + // 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(1, pendingOutboundLimit+2500)} + mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(pendingOutboundLimit + 500)} client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} _, _, err := client.GetAllPendingOutbounds(ctx) @@ -1211,102 +1212,13 @@ func TestClient_GetAllPendingOutbounds_CoversTheWholeSet(t *testing.T) { assert.Contains(t, logBuf.String(), "exceeds one request") }) - t.Run("quiet when the set is fully covered", func(t *testing.T) { + t.Run("quiet when the set fits", func(t *testing.T) { var logBuf bytes.Buffer - mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(2, 2)} + 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 the request limit") - }) -} - -// pendingResp builds a response reporting `total` as the size of the pending set. -func pendingResp(total uint64) *uexecutortypes.QueryAllPendingOutboundsResponse { - return &uexecutortypes.QueryAllPendingOutboundsResponse{ - Entries: []*uexecutortypes.PendingOutboundEntry{{OutboundId: "ob-1"}}, - Outbounds: []*uexecutortypes.OutboundTx{{Id: "ob-1"}}, - Pagination: &query.PageResponse{Total: total}, - } -} - -// New outbounds only arrive at the newest end, so that read is what must never -// be crowded out. The backlog sweep exists only for a set too large to fetch at -// once, and reading it at a fixed offset would re-return the same prefix forever. -func TestClient_GetAllPendingOutbounds_ReadStrategy(t *testing.T) { - ctx := context.Background() - - read := func(m *mockUExecutorQueryClient) (bool, uint64) { - p := m.lastPendingReq.Pagination - return p.Reverse, p.Offset - } - - t.Run("set fits in one request: always newest-first, never sweeps", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(9)} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - for i := 0; i < 4; i++ { - _, _, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - rev, off := read(mockClient) - assert.True(t, rev, "poll %d should read newest-first", i) - assert.Zero(t, off, "poll %d should not sweep a set that fits", i) - } - }) - - t.Run("set too large: newest-first every other poll, backlog advances", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(3500)} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - var got [][2]any - for i := 0; i < 6; i++ { - _, _, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - rev, off := read(mockClient) - got = append(got, [2]any{rev, off}) - } - - // poll 1 has no prior total, so it starts newest-first and the sweep - // engages once the size is known. - assert.Equal(t, [2]any{true, uint64(0)}, got[0]) - assert.Equal(t, [2]any{false, uint64(0)}, got[1], "backlog sweep starts at the oldest end") - assert.Equal(t, [2]any{true, uint64(0)}, got[2], "newest rows read every other poll") - assert.Equal(t, [2]any{false, uint64(1000)}, got[3], "sweep advances") - assert.Equal(t, [2]any{true, uint64(0)}, got[4]) - assert.Equal(t, [2]any{false, uint64(2000)}, got[5], "and keeps advancing") - }) - - t.Run("sweep wraps rather than running off the end", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(2500)} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - var offsets []uint64 - for i := 0; i < 10; i++ { - _, _, err := client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - if rev, off := read(mockClient); !rev { - offsets = append(offsets, off) - } - } - assert.Equal(t, []uint64{0, 1000, 2000, 0, 1000}, offsets, "covers the set then starts over") - }) - - t.Run("sweep stops once the set shrinks to fit", func(t *testing.T) { - mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: pendingResp(3500)} - client := &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{mockClient}} - - _, _, err := client.GetAllPendingOutbounds(ctx) // learns the size - require.NoError(t, err) - mockClient.allPendingOutboundsResp = pendingResp(12) // backlog cleared - - _, _, err = client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - _, _, err = client.GetAllPendingOutbounds(ctx) - require.NoError(t, err) - - rev, off := read(mockClient) - assert.True(t, rev) - assert.Zero(t, off, "no sweep needed once one request covers the set") + assert.NotContains(t, logBuf.String(), "exceeds one request") }) }