#5571: fixed rpc ledger backend to retry requests if request duration limit errors are returned by rpc

Author: Shawn Reuland

ingest/ledgerbackend/rpc_backend.go

@@ -9,13 +9,16 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/creachadair/jrpc2"
 	"github.com/stellar/go/xdr"
 	rpc "github.com/stellar/stellar-rpc/client"
 	"github.com/stellar/stellar-rpc/protocol"
 )
 
 const rpcBackendDefaultBufferSize uint32 = 10
 const rpcBackendDefaultWaitIntervalSeconds uint32 = 2
+const rpcErrRequestExceededProcessingLimitThreshold = -32001
+const rpcErrFailToProcessDueToInternalIssue = -32003
 
 type RPCLedgerMissingError struct {
 	Sequence uint32
@@ -250,6 +253,16 @@ func (b *RPCLedgerBackend) getBufferedLedger(ctx context.Context, sequence uint3
 
 	ledgers, err := b.client.GetLedgers(ctx, req)
 	if err != nil {
+		var rpcErr jrpc2.Error
+		if errors.As(err, &rpcErr) &&
+			// rpc request duration limits may trigger these errors, it is acceptable to retry
+			(rpcErr.Code == rpcErrRequestExceededProcessingLimitThreshold ||
+				rpcErr.Code == rpcErrFailToProcessDueToInternalIssue) {
+			return xdr.LedgerCloseMeta{}, &RPCLedgerBeyondLatestError{
+				Sequence:     sequence,
+				LatestLedger: 0,
+			}
+		}
 		return xdr.LedgerCloseMeta{}, fmt.Errorf("failed to get ledgers starting from %d: %w", sequence, err)
 	}
 

ingest/ledgerbackend/rpc_backend_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/creachadair/jrpc2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 
@@ -210,6 +211,73 @@ func TestGetLedgerBeyondLatest(t *testing.T) {
 
 }
 
+func TestGetLedgerBeyondLatestBasedOnDurationLimitErr(t *testing.T) {
+	rpcBackend, mockClient := setupRPCTest(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	requestedSequence := uint32(100)
+
+	rpcGetLedgersRequest := protocol.GetLedgersRequest{
+		StartLedger: requestedSequence,
+		Pagination: &protocol.LedgerPaginationOptions{
+			Limit: uint(rpcBackendDefaultBufferSize),
+		},
+	}
+	// Setup err responses indicating rpc request elapsed beyond request duration
+	// triggering specific error codes related to request duration limits which rpc may emit
+	exceededProcessingLimitThreshold := jrpc2.Error{
+		Code: -32001,
+	}
+
+	failToProcessDueToInternalIssue := jrpc2.Error{
+		Code: -32003,
+	}
+
+	// called by PrepareRange
+	mockClient.On("GetLedgers", ctx, rpcGetLedgersRequest).Return(protocol.GetLedgersResponse{}, exceededProcessingLimitThreshold).Once()
+	// called by GetLedger first time
+	mockClient.On("GetLedgers", ctx, rpcGetLedgersRequest).Return(protocol.GetLedgersResponse{}, failToProcessDueToInternalIssue).Once()
+
+	// Setup second call to return the requested ledger
+	lcm := xdr.LedgerCloseMeta{
+		V: 0,
+		V0: &xdr.LedgerCloseMetaV0{
+			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
+				Header: xdr.LedgerHeader{
+					LedgerSeq: xdr.Uint32(requestedSequence),
+				},
+			},
+		},
+	}
+	encodedLCM, err := xdr.MarshalBase64(lcm)
+	assert.NoError(t, err)
+
+	secondResponse := protocol.GetLedgersResponse{
+		LatestLedger: requestedSequence,
+		Ledgers: []protocol.LedgerInfo{
+			{
+				Sequence:       requestedSequence,
+				LedgerMetadata: encodedLCM,
+			},
+		},
+	}
+	// called by GetLedger second time
+	mockClient.On("GetLedgers", ctx, rpcGetLedgersRequest).Return(secondResponse, nil).Once()
+
+	preparedRange := Range{from: requestedSequence, to: requestedSequence + 10, bounded: true}
+	rpcBackend.PrepareRange(ctx, preparedRange)
+
+	startTime := time.Now()
+	actualLCM, err := rpcBackend.GetLedger(ctx, requestedSequence)
+	duration := time.Since(startTime)
+
+	assert.NoError(t, err)
+	assert.Equal(t, requestedSequence, uint32(actualLCM.V0.LedgerHeader.Header.LedgerSeq))
+
+	// Verify timing - GetLedger should have waited one interval and then refetched ledgers from rpc on second call
+	assert.GreaterOrEqual(t, duration.Seconds(), float64(rpcBackendDefaultWaitIntervalSeconds))
+}
+
 func TestGetLedgerContextTimeout(t *testing.T) {
 	rpcBackend, mockClient := setupRPCTest(t)
 	sequence := uint32(100)