937 follow-ups: getTransactions cursor guards, getHealth first-commit gate, snapshot gauge, method-table completeness test

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+### Fixed
+* **`getLedgers` no longer rejects its own cursor at the tip.** This changes wire behavior for BOTH rpcv1 and rpcv2 — the handler is shared. Resending the cursor of the last page used to fail with `-32602` ("cursor must be between the oldest ledger ... and the latest ledger ...") until the next ledger closed — the same code a malformed cursor gets, so a poller could not tell "wait and retry" from "bad cursor". A cursor at or past the tip now returns an empty page with the cursor echoed back. A cursor below the oldest ledger still errors (that data is gone), and an explicit `startLedger` above the tip still errors (only the server-issued token gets the echo).
+* **`getTransactions` cursor no longer breaks a caught-up poller.** This changes wire behavior for BOTH rpcv1 and rpcv2 — the handler is shared. Two long-standing bugs, one fix: the returned `cursor` is now never below the request's cursor; at the tip, the request's cursor is echoed back. Before this fix: (1) a cursor at or above the node's latest ledger returned the literal cursor `"0"`, and resending it failed every poll with `-32602` — the client was stuck until it discarded its cursor (reported in [#745](https://github.com/stellar/stellar-rpc/issues/745)); (2) a cursor at a fully-consumed ledger returned a cursor pointing back at that ledger's start, and the next poll re-delivered all of its transactions as duplicates. Clients need no changes: the returned cursor now always does what the docs promise — fetch what comes next.
+
 ### Changed
 * `getTransactions` now scans at most 10,000 ledgers per request. On a sparse range a page can come back short — or even empty — while still carrying a `cursor`, so a response shorter than `limit` no longer implies end-of-data. Every response carries a `cursor`; to tell a sparse scan window from the tip, compare the cursor's ledger part against `latestLedger` and keep paging while it is below ([#908](https://github.com/stellar/stellar-rpc/pull/908)).
 * `getEvents` now rejects malformed contract IDs in filters with `-32602`: a `C…` string with a valid checksum but wrong-length payload used to decode and silently match nothing, and now errors under the SDK's stricter SEP-23 strkey parsing ([#908](https://github.com/stellar/stellar-rpc/pull/908)).

cmd/stellar-rpc/internal/methods/get_ledgers.go

@@ -93,6 +93,21 @@ func (h ledgersHandler) getLedgers(
 		}
 	}
 
+	// A caught-up poller's cursor points at or past the tip. Echo it back on
+	// an empty page instead of rejecting the server's own token. An explicit
+	// startLedger above the tip stays an error (Validate above rejects it).
+	if request.Pagination != nil && request.Pagination.Cursor != "" &&
+		start > availableLedgerRange.LastLedger {
+		return protocol.GetLedgersResponse{
+			Ledgers:               []protocol.LedgerInfo{},
+			LatestLedger:          ledgerRange.LastLedger.Sequence,
+			LatestLedgerCloseTime: ledgerRange.LastLedger.CloseTime,
+			OldestLedger:          ledgerRange.FirstLedger.Sequence,
+			OldestLedgerCloseTime: ledgerRange.FirstLedger.CloseTime,
+			Cursor:                request.Pagination.Cursor,
+		}, nil
+	}
+
 	end := start + uint32(limit) - 1 //nolint:gosec
 	ledgers, err := h.fetchLedgers(ctx, start, end, request.Format, readTx, ledgerRange.ToLedgerSeqRange())
 	if err != nil {
@@ -151,13 +166,16 @@ func (h ledgersHandler) parseCursor(cursor string, ledgerRange protocol.LedgerSe
 		return 0, err
 	}
 
+	// Only the lower bound is an error: below the oldest ledger is data the
+	// node no longer has. At or past the tip is a caught-up poller, answered
+	// with an empty page by getLedgers. The +1 wraps a max-uint32 cursor to
+	// start 0, which this check also catches.
 	start := uint32(cursorInt) + 1
-	if !protocol.IsLedgerWithinRange(start, ledgerRange) {
+	if start < ledgerRange.FirstLedger {
 		return 0, fmt.Errorf(
-			"cursor ('%s') must be between the oldest ledger: %d and the latest ledger: %d for this rpc instance",
+			"cursor ('%s') must be at or above the oldest ledger: %d for this rpc instance",
 			cursor,
 			ledgerRange.FirstLedger,
-			ledgerRange.LastLedger,
 		)
 	}
 

cmd/stellar-rpc/internal/methods/get_ledgers_test.go

@@ -225,7 +225,36 @@ func TestGetLedgers_NoLedgers(t *testing.T) {
 	assert.Contains(t, err.Error(), "[-32603] DB is empty")
 }
 
-func TestGetLedgers_CursorGreaterThanLatestLedger(t *testing.T) {
+func TestGetLedgers_CaughtUpCursorIsEchoed(t *testing.T) {
+	cursors := map[string]string{
+		"at the tip":    "10",
+		"above the tip": "15",
+	}
+	for name, cursor := range cursors {
+		t.Run(name, func(t *testing.T) {
+			testDB := setupTestDB(t, 10)
+			handler := ledgersHandler{
+				ledgerReader: sqlitedb.NewLedgerReader(testDB),
+				maxLimit:     100,
+				defaultLimit: 5,
+			}
+
+			request := protocol.GetLedgersRequest{
+				Pagination: &protocol.LedgerPaginationOptions{
+					Cursor: cursor,
+				},
+			}
+
+			response, err := handler.getLedgers(context.TODO(), request)
+			require.NoError(t, err)
+			assert.Empty(t, response.Ledgers)
+			assert.Equal(t, cursor, response.Cursor)
+			assert.Equal(t, uint32(10), response.LatestLedger)
+		})
+	}
+}
+
+func TestGetLedgers_MaxUint32CursorIsRejected(t *testing.T) {
 	testDB := setupTestDB(t, 10)
 	handler := ledgersHandler{
 		ledgerReader: sqlitedb.NewLedgerReader(testDB),
@@ -235,13 +264,13 @@ func TestGetLedgers_CursorGreaterThanLatestLedger(t *testing.T) {
 
 	request := protocol.GetLedgersRequest{
 		Pagination: &protocol.LedgerPaginationOptions{
-			Cursor: "15",
+			Cursor: "4294967295",
 		},
 	}
 
 	_, err := handler.getLedgers(context.TODO(), request)
 	require.Error(t, err)
-	assert.Contains(t, err.Error(), "cursor ('15') must be between")
+	assert.Contains(t, err.Error(), "must be at or above the oldest ledger")
 }
 
 func BenchmarkGetLedgers(b *testing.B) {

cmd/stellar-rpc/internal/methods/get_transactions.go

@@ -36,27 +36,33 @@ func uint32ToInt32(value uint32, fieldName string) (int32, error) {
 	return int32(parsed), nil
 }
 
-// initializePagination sets the pagination limit and cursor
-func (h transactionsRPCHandler) initializePagination(request protocol.GetTransactionsRequest) (toid.ID, uint, error) {
+// initializePagination sets the pagination limit and cursor. The second
+// return value is the request's own cursor, nil when the request has none.
+func (h transactionsRPCHandler) initializePagination(
+	request protocol.GetTransactionsRequest,
+) (toid.ID, *toid.ID, uint, error) {
 	startLedger, err := uint32ToInt32(request.StartLedger, "startLedger")
 	if err != nil {
-		return toid.ID{}, 0, &jrpc2.Error{
+		return toid.ID{}, nil, 0, &jrpc2.Error{
 			Code:    jrpc2.InvalidParams,
 			Message: err.Error(),
 		}
 	}
 	start := toid.New(startLedger, 1, 1)
 	limit := h.defaultLimit
+	var requestCursor *toid.ID
 	if request.Pagination != nil {
 		if request.Pagination.Cursor != "" {
 			cursorInt, err := strconv.ParseInt(request.Pagination.Cursor, 10, 64)
 			if err != nil {
-				return toid.ID{}, 0, &jrpc2.Error{
+				return toid.ID{}, nil, 0, &jrpc2.Error{
 					Code:    jrpc2.InvalidParams,
 					Message: err.Error(),
 				}
 			}
-			*start = toid.Parse(cursorInt)
+			parsed := toid.Parse(cursorInt)
+			requestCursor = &parsed
+			*start = parsed
 			// increment tx index because, when paginating,
 			// we start with the item right after the cursor
 			start.TransactionOrder++
@@ -65,7 +71,7 @@ func (h transactionsRPCHandler) initializePagination(request protocol.GetTransac
 			limit = request.Pagination.Limit
 		}
 	}
-	return *start, limit, nil
+	return *start, requestCursor, limit, nil
 }
 
 // fetchLedgerData calls the meta table to fetch the corresponding ledger data.
@@ -245,7 +251,7 @@ func (h transactionsRPCHandler) getTransactionsByLedgerSequence(ctx context.Cont
 		}
 	}
 
-	start, limit, err := h.initializePagination(request)
+	start, requestCursor, limit, err := h.initializePagination(request)
 	if err != nil {
 		return protocol.GetTransactionsResponse{}, err
 	}
@@ -280,6 +286,14 @@ func (h transactionsRPCHandler) getTransactionsByLedgerSequence(ctx context.Cont
 		}
 	}
 
+	// A caught-up poller's cursor points at or past the tip. The walk then
+	// produces nothing and leaves the cursor below the request's own — at the
+	// zero value, or at the consumed ledger's start. Echo the request's cursor
+	// instead: the returned token must always fetch what comes next (#745).
+	if requestCursor != nil && cursor.ToInt64() < requestCursor.ToInt64() {
+		cursor = requestCursor
+	}
+
 	return protocol.GetTransactionsResponse{
 		Transactions:          txns,
 		LatestLedger:          ledgerRange.LastLedger.Sequence,

cmd/stellar-rpc/internal/methods/get_transactions_test.go

@@ -149,6 +149,36 @@ func TestGetTransactions_CustomLimitAndCursor(t *testing.T) {
 	assert.Equal(t, uint32(3), response.Transactions[2].Ledger)
 }
 
+func TestGetTransactions_CaughtUpCursorIsEchoed(t *testing.T) {
+	cursors := map[string]string{
+		"above the tip":          toid.New(15, 1, 1).String(),
+		"at the consumed tip":    toid.New(10, 2, 1).String(),
+		"past the tip's last tx": toid.New(10, 5, 1).String(),
+	}
+	for name, cursor := range cursors {
+		t.Run(name, func(t *testing.T) {
+			testDB := setupDB(t, 10, 0)
+			handler := transactionsRPCHandler{
+				ledgerReader:      sqlitedb.NewLedgerReader(testDB),
+				maxLimit:          100,
+				defaultLimit:      10,
+				networkPassphrase: NetworkPassphrase,
+			}
+
+			request := protocol.GetTransactionsRequest{
+				Pagination: &protocol.LedgerPaginationOptions{
+					Cursor: cursor,
+				},
+			}
+
+			response, err := handler.getTransactionsByLedgerSequence(context.TODO(), request)
+			require.NoError(t, err)
+			assert.Empty(t, response.Transactions)
+			assert.Equal(t, cursor, response.Cursor)
+		})
+	}
+}
+
 func TestGetTransactions_InvalidStartLedger(t *testing.T) {
 	testDB := setupDB(t, 3, 0)
 	handler := transactionsRPCHandler{

cmd/stellar-rpc/internal/rpcv2/jsonrpc.go

@@ -80,6 +80,9 @@ func newJSONRPCHandler(cfg config.Config, p handlerParams) jsonrpc.Handler {
 	specs = limitsByMethod(m).Apply(specs)
 	for i := range specs {
 		specs[i].Handler = wrapAdapterRequest(specs[i].Handler, p.registry)
+		if specs[i].MethodName == protocol.GetHealthMethodName {
+			specs[i].Handler = gateHealthOnFirstCommit(specs[i].Handler, p.registry)
+		}
 	}
 
 	return jsonrpc.NewHandler(jsonrpc.Params{
@@ -155,6 +158,22 @@ func wrapAdapterRequest(h jrpc2.Handler, registry *query.Registry) jrpc2.Handler
 	}
 }
 
+// gateHealthOnFirstCommit fails getHealth until this run commits a ledger.
+// Close times survive restarts in the durable stores, so freshness alone
+// cannot tell a working node from one whose ingestion never started. v1
+// keeps the shared behavior.
+func gateHealthOnFirstCommit(h jrpc2.Handler, registry *query.Registry) jrpc2.Handler {
+	return func(ctx context.Context, req *jrpc2.Request) (any, error) {
+		if !registry.HasCommittedSinceBoot() {
+			return nil, &jrpc2.Error{
+				Code:    jrpc2.InternalError,
+				Message: "ingestion has not committed a ledger since this process started",
+			}
+		}
+		return h(ctx, req)
+	}
+}
+
 // graceMargin is the slack deriveLifecycleGrace adds on top of the longest
 // request timeout. It covers the gap between a request's deadline firing and
 // its handler goroutine actually stopping: the duration limiter answers the

cmd/stellar-rpc/internal/rpcv2/jsonrpc_test.go

@@ -112,6 +112,26 @@ func TestJSONRPCHandler_HealthyOverFreshRegistryStamp(t *testing.T) {
 	assert.Equal(t, "healthy", result.Status)
 }
 
+func TestJSONRPCHandler_HealthGatedUntilFirstCommit(t *testing.T) {
+	r := seedServingRegistry(t)
+	r.SeedLatestAtBoot(chunk.FirstLedgerSeq, time.Now().Unix())
+	url := newTestRPCServer(t, r)
+
+	out := rpcv2test.PostRPC(t, url, "getHealth", `{}`)
+	require.NotNil(t, out.Error)
+	assert.EqualValues(t, jrpc2.InternalError, out.Error.Code)
+	assert.Contains(t, out.Error.Message, "since this process started")
+
+	r.SetLatestLedger(chunk.FirstLedgerSeq+1, time.Now().Unix())
+	out = rpcv2test.PostRPC(t, url, "getHealth", `{}`)
+	require.Nil(t, out.Error)
+	var result struct {
+		Status string `json:"status"`
+	}
+	require.NoError(t, json.Unmarshal(out.Result, &result))
+	assert.Equal(t, "healthy", result.Status)
+}
+
 func TestWrapAdapterRequest_PanicReleasesSharedView(t *testing.T) {
 	logger, buf := capturingLogger()
 	cat, _ := rpcv2test.OpenTestCatalogWith(t, testCPI, logger)

cmd/stellar-rpc/internal/rpcv2/methods_completeness_test.go

@@ -0,0 +1,57 @@
+package rpcv2
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/config"
+)
+
+// configMethodNames maps each MethodsConfig method field's Go name to its wire
+// name (the toml tag). The two methods-wide default-tier fields are pointers,
+// not structs, so the kind check skips them.
+func configMethodNames(t *testing.T) map[string]string {
+	t.Helper()
+	typ := reflect.TypeFor[config.MethodsConfig]()
+	names := map[string]string{}
+	for field := range typ.Fields() {
+		if field.Type.Kind() != reflect.Struct {
+			continue
+		}
+		wireName, _, _ := strings.Cut(field.Tag.Get("toml"), ",")
+		require.NotEmpty(t, wireName, "method field %s needs a toml tag", field.Name)
+		names[field.Name] = wireName
+	}
+	return names
+}
+
+func TestLimitsByMethod_CoversEveryConfiguredMethod(t *testing.T) {
+	limits := limitsByMethod(validCfg(1, 1, "genesis").Service.Methods)
+
+	wireNames := configMethodNames(t)
+	for _, name := range wireNames {
+		assert.Contains(t, limits, name)
+	}
+	assert.Len(t, limits, len(wireNames),
+		"limitsByMethod has a key with no MethodsConfig field behind it")
+}
+
+func TestValidateService_ChecksEveryConfiguredMethod(t *testing.T) {
+	for fieldName, wireName := range configMethodNames(t) {
+		t.Run(wireName, func(t *testing.T) {
+			cfg := validCfg(1, 1, "genesis")
+			methodField := reflect.ValueOf(&cfg.Service.Methods).Elem().FieldByName(fieldName)
+			zero := uint(0)
+			methodField.FieldByName("QueueLimit").Set(reflect.ValueOf(&zero))
+
+			err := validateService(cfg.Service)
+			require.Error(t, err,
+				"validateService's method list is missing %s", wireName)
+			assert.Contains(t, err.Error(), wireName)
+		})
+	}
+}

cmd/stellar-rpc/internal/rpcv2/observability/observability.go

@@ -180,6 +180,11 @@ func NewPrometheusMetrics(registry *prometheus.Registry, namespace string) *Prom
 			"cold ledger packs whose file was gone on first read "+
 				"(routing only opens packs the catalog snapshot holds; any count is an alarm)",
 			ledger.MissingPackOpens),
+		prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+			Namespace: namespace, Subsystem: subsystem, Name: "open_snapshots",
+			Help: "RocksDB snapshots currently held, across all stores " +
+				"(request-scoped, so a floor that stops returning to zero is a leaked read view)",
+		}, func() float64 { return float64(rocksdb.OpenSnapshots()) }),
 	)
 	return m
 }

cmd/stellar-rpc/internal/rpcv2/observability/observability_test.go

@@ -75,6 +75,9 @@ func TestPrometheusMetrics_RegistersAndRecords(t *testing.T) {
 	assert.InDelta(t, float64(3), values["test_ns_fullhistory_streaming_discarded_hot_chunks_total"], 0)
 	assert.InDelta(t, float64(2), values["test_ns_fullhistory_streaming_pruned_artifacts_total"], 0)
 
+	_, exported := values["test_ns_fullhistory_streaming_open_snapshots"]
+	assert.True(t, exported, "open_snapshots gauge must be registered")
+
 	// Phase-duration histogram saw backfill_pass + freeze + rebuild + discard + prune = 5 observations.
 	assert.Equal(t, uint64(5), counts["test_ns_fullhistory_streaming_phase_duration_seconds"])
 }

cmd/stellar-rpc/internal/rpcv2/query/registry.go

@@ -42,6 +42,9 @@ type Registry struct {
 	// (ReadView.LatestLedger / LatestCloseTime), never this live value.
 	latest atomic.Pointer[ledgerStamp]
 
+	// bootSeq is the latest ledger found at startup; set once by SeedLatestAtBoot.
+	bootSeq uint32
+
 	// oldest is a read-through cache of the retention floor's first ledger and
 	// its close time. Views populate it after a fallback point read
 	// (ReadView.RecordOldestCloseTime); readers trust it only while its seq
@@ -118,7 +121,7 @@ func OpenRegistry(
 	}
 	r.PublishHandle(live.ChunkID(), live)
 	// The catalog has no close times, so the seed stamp starts at 0 (unknown).
-	r.SetLatestLedger(lastCommitted, 0)
+	r.SeedLatestAtBoot(lastCommitted, 0)
 	return r, nil
 }
 
@@ -145,6 +148,19 @@ func (r *Registry) SetLatestLedger(seq uint32, closeTimeUnix int64) {
 	r.latest.Store(&ledgerStamp{seq: seq, closeTime: closeTimeUnix})
 }
 
+// SeedLatestAtBoot is SetLatestLedger plus a record of seq as this process's
+// boot value. OpenRegistry calls it once at startup.
+func (r *Registry) SeedLatestAtBoot(seq uint32, closeTimeUnix int64) {
+	r.bootSeq = seq
+	r.SetLatestLedger(seq, closeTimeUnix)
+}
+
+// HasCommittedSinceBoot reports whether ingestion advanced the latest ledger
+// past the boot value.
+func (r *Registry) HasCommittedSinceBoot() bool {
+	return r.latest.Load().seq > r.bootSeq
+}
+
 // LatestLedger returns the live latest ledger. Queries do not call this — they
 // read the frozen ReadView.LatestLedger captured at acquisition (see the
 // latest field).