rpcv2: getEvents read-path performance across both tiers

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

@@ -5,6 +5,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/packfile"
 	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/query"
 	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/rocksdb"
 	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/stores/ledger"
@@ -153,10 +154,10 @@ func NewPrometheusMetrics(registry *prometheus.Registry, namespace string) *Prom
 		}, []string{"phase"}),
 	}
 
-	// Three serving-invariant counters are tallied where their condition is
-	// detected — deep in the store and routing packages, below any metrics
-	// plumbing — as package-level atomics. CounterFuncs export them; each
-	// should flatline at zero, so one alert rule per family is "rate > 0".
+	// Four serving-invariant counters are tallied where their condition is
+	// detected — deep in the store, routing and packfile packages, below any
+	// metrics plumbing — as package-level atomics. CounterFuncs export them;
+	// each should flatline at zero, so one alert rule per family is "rate > 0".
 	counterFunc := func(name, help string, read func() uint64) prometheus.CounterFunc {
 		return prometheus.NewCounterFunc(prometheus.CounterOpts{
 			Namespace: namespace, Subsystem: subsystem, Name: name, Help: help,
@@ -180,6 +181,10 @@ 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),
+		counterFunc("pooled_buffer_cap_skips_total",
+			"pooled packfile buffers dropped for exceeding a pool's capacity cap "+
+				"(the pool drains and every open allocates afresh; any count means a cap wants raising)",
+			packfile.PoolCapSkips),
 		prometheus.NewGaugeFunc(prometheus.GaugeOpts{
 			Namespace: namespace, Subsystem: subsystem, Name: "open_snapshots",
 			Help: "RocksDB snapshots currently held, across all stores " +

cmd/stellar-rpc/internal/rpcv2/packfile/index.go

@@ -79,7 +79,8 @@ func decodeIndex(buf []byte, recordCount int, indexSize int, indexBase int64) ([
 	// (width, min) is at its tail, so intpack.DecodeGroup naturally reads from the end of
 	// its window. Iterating backward lets us shrink the window after each group.
 	groupCount := (recordCount + groupSize - 1) / groupSize
-	deltas := make([]uint32, recordCount)
+	deltas := getScratch(recordCount)
+	defer putScratch(deltas)
 	pos := payloadLen
 
 	for g := groupCount - 1; g >= 0; g-- {
@@ -102,8 +103,9 @@ func decodeIndex(buf []byte, recordCount int, indexSize int, indexBase int64) ([
 		return nil, fmt.Errorf("%w: index has %d unconsumed bytes after decoding all groups", ErrCorrupt, pos)
 	}
 
-	// Forward prefix-sum to build absolute offsets from deltas.
-	offsets := make([]int64, recordCount+1)
+	// Forward prefix-sum from deltas to absolute offsets. Every entry of the
+	// pooled table is overwritten below, the sentinel included.
+	offsets := getOffsets(recordCount + 1)
 
 	offset := int64(0)
 	for i, d := range deltas {
@@ -113,6 +115,7 @@ func decodeIndex(buf []byte, recordCount int, indexSize int, indexBase int64) ([
 
 	// Structural sanity check: running sum must arrive at indexBase.
 	if offset != indexBase {
+		putOffsets(offsets)
 		return nil, fmt.Errorf("%w: final offset %d != indexBase %d", ErrCorrupt, offset, indexBase)
 	}
 

cmd/stellar-rpc/internal/rpcv2/packfile/pools.go

@@ -0,0 +1,124 @@
+package packfile
+
+// Pools for the open path. Cold packfiles are opened per request, and every
+// open decodes an offset table, with FOR-decode scratch, out of read buffers;
+// these pools recycle that memory. Contents are fully rewritten before use,
+// and a pooled buffer is either dead before its function returns or
+// reader-private until Close returns it.
+//
+// Puts are capped so one pathological file cannot pin a huge array. A cap
+// must sit above what a legitimate packfile needs: crossing it skips the Put,
+// the pool drains, and every open allocates afresh. capSkips counts those
+// skips so the drain is visible.
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+// maxPooledOffsets caps the decoded offset table (recordCount+1 int64s) a Put
+// may retain. Chunk geometry, which this package cannot import, sets what it
+// must exceed: the ledger pack holds one ledger per record, 10,001 entries;
+// events.pack and index.pack hold 128 items per record, so 1<<20 entries
+// covers about 134M events or index terms per chunk, against roughly 600K
+// terms in a production chunk today. Raise it if the geometry grows.
+const maxPooledOffsets = 1 << 20 // entries (8 MiB backing array)
+
+const (
+	// maxPooledScratch tracks maxPooledOffsets: the FOR-decode scratch is
+	// recordCount uint32s to the offset table's recordCount+1 int64s.
+	maxPooledScratch = 1 << 20 // entries (4 MiB backing array)
+	maxPooledOpenBuf = 4 << 20 // bytes
+)
+
+//nolint:gochecknoglobals // process-wide pools, like recordWorkspacePool
+var (
+	offsetsPool sync.Pool // *[]int64
+	scratchPool sync.Pool // *[]uint32
+	openBufPool sync.Pool // *[]byte
+)
+
+// capSkips counts Puts dropped for exceeding a cap, across all three pools.
+// A count that climbs with the open rate means the pools have stopped
+// recycling. Exported through PoolCapSkips.
+//
+//nolint:gochecknoglobals // one tally across process-wide pools; read-only outside this file
+var capSkips atomic.Uint64
+
+// PoolCapSkips returns the process-wide count of pooled buffers dropped for
+// exceeding a capacity cap. Zero is the only healthy value; a climbing count
+// means a cap wants raising.
+func PoolCapSkips() uint64 { return capSkips.Load() }
+
+// On a size miss the pooled buffer is returned before a larger one is
+// allocated: Get already removed it, and returning it is what keeps a run of
+// growing opens from draining the pool. It is under its cap, so it never
+// counts as a skip.
+
+func getOffsets(n int) []int64 {
+	if p, _ := offsetsPool.Get().(*[]int64); p != nil {
+		if cap(*p) >= n {
+			return (*p)[:n]
+		}
+		putOffsets(*p)
+	}
+	return make([]int64, n)
+}
+
+// putOffsets recycles a decoded offset table. The caller must guarantee no
+// live reference remains; see Reader.Close for the in-flight handshake.
+func putOffsets(s []int64) {
+	if cap(s) == 0 {
+		return
+	}
+	if cap(s) > maxPooledOffsets {
+		capSkips.Add(1)
+		return
+	}
+	s = s[:0]
+	offsetsPool.Put(&s)
+}
+
+func getScratch(n int) []uint32 {
+	if p, _ := scratchPool.Get().(*[]uint32); p != nil {
+		if cap(*p) >= n {
+			return (*p)[:n]
+		}
+		putScratch(*p)
+	}
+	return make([]uint32, n)
+}
+
+func putScratch(s []uint32) {
+	if cap(s) == 0 {
+		return
+	}
+	if cap(s) > maxPooledScratch {
+		capSkips.Add(1)
+		return
+	}
+	s = s[:0]
+	scratchPool.Put(&s)
+}
+
+func getOpenBuf(n int) []byte {
+	if p, _ := openBufPool.Get().(*[]byte); p != nil {
+		if cap(*p) >= n {
+			return (*p)[:n]
+		}
+		putOpenBuf(*p)
+	}
+	return make([]byte, n)
+}
+
+func putOpenBuf(s []byte) {
+	if cap(s) == 0 {
+		return
+	}
+	if cap(s) > maxPooledOpenBuf {
+		capSkips.Add(1)
+		return
+	}
+	s = s[:0]
+	openBufPool.Put(&s)
+}

cmd/stellar-rpc/internal/rpcv2/packfile/pools_test.go

@@ -0,0 +1,120 @@
+package packfile
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// poolItems builds a deterministic item set spanning several records.
+func poolItems(n int) [][]byte {
+	items := make([][]byte, n)
+	for i := range items {
+		items[i] = bytes.Repeat([]byte{byte(i), byte(i >> 8)}, 64)
+	}
+	return items
+}
+
+// Several open-read-close cycles over distinct files, so a recycled buffer
+// that leaked one file's decode into another's would fail the byte checks.
+func TestReaderPoolReuseKeepsReadsCorrect(t *testing.T) {
+	for cycle := range 4 {
+		// Different item counts reuse recycled arrays at different lengths.
+		items := poolItems(300 + 40*cycle)
+		path := writeTestPackfile(t, items, WriterOptions{ItemsPerRecord: 16})
+		r := Open(path, ReaderOptions{})
+		for i, want := range items {
+			require.NoError(t, r.ReadItem(i, func(got []byte) error {
+				if !bytes.Equal(got, want) {
+					return fmt.Errorf("cycle %d item %d mismatch", cycle, i)
+				}
+				return nil
+			}))
+		}
+		require.NoError(t, r.Close())
+	}
+}
+
+// A read beginning after Close reports os.ErrClosed rather than touching
+// recycled memory.
+func TestReaderReadAfterCloseFails(t *testing.T) {
+	items := poolItems(64)
+	path := writeTestPackfile(t, items, WriterOptions{ItemsPerRecord: 16})
+	r := Open(path, ReaderOptions{})
+	require.NoError(t, r.ReadItem(0, func([]byte) error { return nil }))
+	require.NoError(t, r.Close())
+
+	err := r.ReadItem(0, func([]byte) error { return nil })
+	require.ErrorIs(t, err, os.ErrClosed)
+	err = r.ReadItems(context.Background(), []int{0}, func(int, []byte) error { return nil })
+	require.ErrorIs(t, err, os.ErrClosed)
+	for _, err := range r.ReadRange(0, 1) {
+		require.ErrorIs(t, err, os.ErrClosed)
+	}
+}
+
+// Close racing an in-flight read, a caller contract violation, must leave the
+// offsets to the garbage collector rather than recycle them under the reader.
+func TestReaderCloseDuringReadDoesNotRecycle(t *testing.T) {
+	items := poolItems(256)
+	path := writeTestPackfile(t, items, WriterOptions{ItemsPerRecord: 16})
+	r := Open(path, ReaderOptions{})
+
+	parked := make(chan struct{})
+	unpark := make(chan struct{})
+	var closeErr error
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		<-parked
+		closeErr = r.Close()
+		close(unpark)
+	})
+
+	first := true
+	got := make([]byte, 0, len(items[0]))
+	err := r.ReadItems(context.Background(), []int{0}, func(_ int, data []byte) error {
+		got = append(got[:0], data...)
+		if first {
+			first = false
+			close(parked)
+			<-unpark
+		}
+		return nil
+	})
+	wg.Wait()
+	require.NoError(t, closeErr)
+	// Either outcome is allowed; what the handshake forbids is recycled-memory
+	// corruption, which the byte check would catch.
+	if err == nil {
+		require.True(t, bytes.Equal(got, items[0]), "payload corrupted by Close during read")
+	} else {
+		require.ErrorIs(t, err, os.ErrClosed)
+	}
+}
+
+// The counter must count exactly the Puts a cap dropped, on every pool, and
+// nothing for kept buffers or ignored empty slices. It is process-wide, so
+// the assertions are on the delta.
+func TestPoolCapSkipsCountsDroppedPuts(t *testing.T) {
+	before := PoolCapSkips()
+
+	putOffsets(make([]int64, 0, maxPooledOffsets))
+	putScratch(make([]uint32, 0, maxPooledScratch))
+	putOpenBuf(make([]byte, 0, maxPooledOpenBuf))
+	putOffsets(nil)
+	putScratch(nil)
+	putOpenBuf(nil)
+	require.Equal(t, before, PoolCapSkips(),
+		"a Put at the cap is pooled and an empty one is ignored; neither is a cap skip")
+
+	putOffsets(make([]int64, 0, maxPooledOffsets+1))
+	putScratch(make([]uint32, 0, maxPooledScratch+1))
+	putOpenBuf(make([]byte, 0, maxPooledOpenBuf+1))
+	require.Equal(t, before+3, PoolCapSkips(),
+		"each pool must count the Put its cap dropped")
+}