fix: carry pending outbound cursor between polls so the page budget costs latency not coverage (F-2026-18817)

Author: Aman035

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,27 +446,36 @@ 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
 		}
 
 		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
 	}
 
+	// 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
 }
 

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")
+}