fix: walk pending outbounds oldest-first instead of reading one page

Author: Aman035

universalClient/pushcore/pushCore.go

@@ -395,64 +395,66 @@ 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.
+// poll; anything beyond it 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
+	// A row costs roughly a kilobyte on the wire, so a page stays well inside
+	// gRPC's 4 MiB default. Asking for the whole set in one request would fail
+	// the call outright once the set grew, taking the poll down entirely.
+	pendingOutboundPageSize = 1000
+	pendingOutboundMaxPages = 5
 
 	chainConfigPageSize = 200
 	chainConfigMaxPages = 20
 )
 
-// GetAllPendingOutbounds retrieves pending outbound transactions from Push Chain.
+// GetAllPendingOutbounds retrieves pending outbound transactions from Push Chain,
+// oldest first, so older work is signed before newer.
 //
-// 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.
+// Walked by offset rather than read as a single page. An outbound leaves the
+// pending set only when a quorum vote terminalizes it, so rows that cannot reach
+// one accumulate at the head of the list; without the walk they would hide every
+// newer outbound behind them, 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.
+// The walk stops at the first short page, so the ordinary case where the whole
+// set fits in one page costs exactly one request.
 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: pendingOutboundLimit, Reverse: true},
-			})
-		},
-		"GetAllPendingOutbounds",
-		c.logger,
+	var (
+		entries   []*uexecutortypes.PendingOutboundEntry
+		outbounds []*uexecutortypes.OutboundTx
 	)
-	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")
+	for page := 0; page < pendingOutboundMaxPages; page++ {
+		offset := uint64(page) * pendingOutboundPageSize
+		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{Offset: offset, Limit: pendingOutboundPageSize},
+				})
+			},
+			"GetAllPendingOutbounds",
+			c.logger,
+		)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		entries = append(entries, resp.Entries...)
+		outbounds = append(outbounds, resp.Outbounds...)
+
+		if len(resp.Entries) < pendingOutboundPageSize {
+			return entries, outbounds, nil
+		}
 	}
 
-	return resp.Entries, resp.Outbounds, nil
+	c.logger.Warn().
+		Int("max_pages", pendingOutboundMaxPages).
+		Int("fetched", len(entries)).
+		Msg("pending outbound page cap reached; the remainder is read on the next poll")
+
+	return entries, outbounds, nil
 }
 
 // createGRPCConnection creates a gRPC connection with appropriate transport security.

universalClient/pushcore/pushCore_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"math/big"
 	"testing"
 
@@ -1048,6 +1049,11 @@ type mockUExecutorQueryClient struct {
 
 	// lastPendingReq records the request so tests can assert the limit sent.
 	lastPendingReq *uexecutortypes.QueryAllPendingOutboundsRequest
+
+	// pendingReqs records every page request of a walk.
+	pendingReqs []*uexecutortypes.QueryAllPendingOutboundsRequest
+	// pendingTotal, when set, makes the mock serve that many rows by offset.
+	pendingTotal int
 }
 
 func (m *mockUExecutorQueryClient) GasPrice(ctx context.Context, req *uexecutortypes.QueryGasPriceRequest, opts ...grpc.CallOption) (*uexecutortypes.QueryGasPriceResponse, error) {
@@ -1075,9 +1081,26 @@ 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
+	m.pendingReqs = append(m.pendingReqs, req)
 	if m.err != nil {
 		return nil, m.err
 	}
+	if m.pendingTotal > 0 {
+		offset := int(req.Pagination.GetOffset())
+		end := offset + int(req.Pagination.GetLimit())
+		if end > m.pendingTotal {
+			end = m.pendingTotal
+		}
+		resp := &uexecutortypes.QueryAllPendingOutboundsResponse{
+			Pagination: &query.PageResponse{Total: uint64(m.pendingTotal)},
+		}
+		for i := offset; i < end; i++ {
+			id := fmt.Sprintf("ob-%d", i)
+			resp.Entries = append(resp.Entries, &uexecutortypes.PendingOutboundEntry{OutboundId: id})
+			resp.Outbounds = append(resp.Outbounds, &uexecutortypes.OutboundTx{Id: id})
+		}
+		return resp, nil
+	}
 	return m.allPendingOutboundsResp, nil
 }
 
@@ -1171,54 +1194,74 @@ func TestClient_GetKeyByID(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) {
+// An outbound leaves the pending set only on a quorum vote, so rows that cannot
+// reach one accumulate at the head of an oldest-first list. The walk is what
+// stops them hiding everything newer.
+func TestClient_GetAllPendingOutbounds_WalksOldestFirst(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},
-		}
+	newClient := func(total int) (*Client, *mockUExecutorQueryClient) {
+		m := &mockUExecutorQueryClient{pendingTotal: total}
+		return &Client{logger: zerolog.Nop(), uexecutorClients: []uexecutortypes.QueryClient{m}}, m
 	}
 
-	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}}
-
+	t.Run("oldest first, never reversed", func(t *testing.T) {
+		client, m := newClient(9)
 		_, _, err := client.GetAllPendingOutbounds(ctx)
 		require.NoError(t, err)
 
-		p := mockClient.lastPendingReq.Pagination
+		p := m.pendingReqs[0].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)
+		assert.False(t, p.Reverse, "older outbounds must be read first")
+		assert.Zero(t, p.Offset)
+		assert.Equal(t, uint64(pendingOutboundPageSize), 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) {
+	// The ordinary case is a set that fits, and it must not cost extra requests.
+	t.Run("a set that fits costs one request", func(t *testing.T) {
+		client, m := newClient(9)
+		entries, _, err := client.GetAllPendingOutbounds(ctx)
+		require.NoError(t, err)
+		assert.Len(t, m.pendingReqs, 1)
+		assert.Len(t, entries, 9)
+	})
+
+	// A stuck prefix must not hide what is behind it.
+	t.Run("walks past a full first page", func(t *testing.T) {
+		client, m := newClient(pendingOutboundPageSize + 250)
+		entries, outbounds, err := client.GetAllPendingOutbounds(ctx)
+		require.NoError(t, err)
+
+		require.Len(t, m.pendingReqs, 2)
+		assert.Equal(t, uint64(0), m.pendingReqs[0].Pagination.GetOffset())
+		assert.Equal(t, uint64(pendingOutboundPageSize), m.pendingReqs[1].Pagination.GetOffset())
+
+		require.Len(t, entries, pendingOutboundPageSize+250)
+		require.Len(t, outbounds, pendingOutboundPageSize+250)
+		assert.Equal(t, "ob-0", entries[0].OutboundId, "oldest first")
+		assert.Equal(t, fmt.Sprintf("ob-%d", pendingOutboundPageSize+249), entries[len(entries)-1].OutboundId)
+	})
+
+	// The cap bounds one poll; the rest is read on the next tick.
+	t.Run("stops at the page cap and says so", func(t *testing.T) {
 		var logBuf bytes.Buffer
-		mockClient := &mockUExecutorQueryClient{allPendingOutboundsResp: resp(pendingOutboundLimit + 500)}
-		client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{mockClient}}
+		m := &mockUExecutorQueryClient{pendingTotal: pendingOutboundPageSize * (pendingOutboundMaxPages + 2)}
+		client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{m}}
 
-		_, _, err := client.GetAllPendingOutbounds(ctx)
+		entries, _, err := client.GetAllPendingOutbounds(ctx)
 		require.NoError(t, err)
-		assert.Contains(t, logBuf.String(), "exceeds one request")
+		assert.Len(t, m.pendingReqs, pendingOutboundMaxPages)
+		assert.Len(t, entries, pendingOutboundPageSize*pendingOutboundMaxPages)
+		assert.Contains(t, logBuf.String(), "page cap reached")
 	})
 
 	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}}
+		m := &mockUExecutorQueryClient{pendingTotal: 9}
+		client := &Client{logger: zerolog.New(&logBuf), uexecutorClients: []uexecutortypes.QueryClient{m}}
 
 		_, _, err := client.GetAllPendingOutbounds(ctx)
 		require.NoError(t, err)
-		assert.NotContains(t, logBuf.String(), "exceeds one request")
+		assert.NotContains(t, logBuf.String(), "page cap reached")
 	})
 }