events: seal the hot index one slab at a time

cmd/stellar-rpc/internal/rpcv2/bench/csvsink.go

@@ -153,11 +153,10 @@ func newCSVSink() *csvSink {
 // percentiles do not add up to a per-ledger total.
 //
 // The production hot loop is one goroutine, so phases arrive as one burst per
-// ledger in hotchunk.Phase order. PhaseExtract starts a burst. PhaseApply ends
-// it and is emitted only on success, so a failed ledger records no
-// ingest_total. Interleaved bursts from several goroutines are safe but sum
-// across ledgers.
-func (s *csvSink) HotPhase(phase hotchunk.Phase, d time.Duration, items int, _ error) {
+// ledger in hotchunk.Phase order. PhaseExtract starts a burst. A PhaseApply
+// without an error ends it, so a failed ledger records no ingest_total.
+// Interleaved bursts from several goroutines are safe but sum across ledgers.
+func (s *csvSink) HotPhase(phase hotchunk.Phase, d time.Duration, items int, err error) {
 	s.observe(fileHot, phase.String(), d, items)
 
 	// Release mu before observe takes it again.
@@ -169,7 +168,7 @@ func (s *csvSink) HotPhase(phase hotchunk.Phase, d time.Duration, items int, _ e
 	total := s.hotBurst
 	s.mu.Unlock()
 
-	if phase == hotchunk.PhaseApply {
+	if phase == hotchunk.PhaseApply && err == nil {
 		s.observe(fileDriver, driverIngestTotal, total, 1)
 	}
 }

cmd/stellar-rpc/internal/rpcv2/bench/csvsink_test.go

@@ -83,11 +83,11 @@ func TestCSVSinkExactOutput(t *testing.T) {
 
 // TestCSVSinkHotIngestTotal drives the ingest_total reconstruction directly:
 // two complete per-ledger HotPhase bursts (extract→ledgers→txhash→events→
-// commit→apply) plus one FAILED burst (extract, then a phase carrying an error,
-// no apply). Only PhaseApply — the terminal, success-only phase — emits an
-// ingest_total sample, so the failed burst contributes nothing; each complete
-// burst contributes one sample (items=1) whose duration is the sum of that
-// burst's phases.
+// commit→apply) plus two FAILED bursts (one ending at a failed commit, one at
+// a failed apply). Only a PhaseApply without an error emits an ingest_total
+// sample, so the failed bursts contribute nothing; each complete burst
+// contributes one sample (items=1) whose duration is the sum of that burst's
+// phases.
 func TestCSVSinkHotIngestTotal(t *testing.T) {
 	sink := newCSVSink()
 
@@ -104,10 +104,13 @@ func TestCSVSinkHotIngestTotal(t *testing.T) {
 	burst(100) // 6*100 + 21 = 621
 	burst(200) // 6*200 + 21 = 1221
 
-	// A failed ledger: extract ran, then PhaseCommit failed — no apply, so no
-	// ingest_total sample.
+	// Failed ledgers: one whose commit failed (no apply), one whose apply
+	// failed. Neither adds an ingest_total sample.
 	sink.HotPhase(hotchunk.PhaseExtract, 10, 0, nil)
 	sink.HotPhase(hotchunk.PhaseCommit, 20, 0, errors.New("commit failed"))
+	sink.HotPhase(hotchunk.PhaseExtract, 10, 0, nil)
+	sink.HotPhase(hotchunk.PhaseCommit, 20, 0, nil)
+	sink.HotPhase(hotchunk.PhaseApply, 30, 0, errors.New("apply failed"))
 
 	outDir := t.TempDir()
 	_, err := sink.writeCSVs(outDir)

cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go

@@ -1276,6 +1276,47 @@ func TestHotService_EmitsEveryPhaseOnSuccess(t *testing.T) {
 	assert.False(t, hadErr, "success path carries no phase error")
 }
 
+// A failed apply reports every phase, the write phases with what landed and
+// the error on PhaseApply.
+func TestHotService_EmitsEveryPhaseOnFailedApply(t *testing.T) {
+	dir := t.TempDir()
+	db, err := hotchunk.Open(dir, chunk.ID(0), hotTestLogger())
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+	// A file where the events index wants its load directory makes every
+	// seal fail.
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "loading"), nil, 0o600))
+
+	sink := &testSink{}
+	svc := NewHotService(db, nil, sink)
+	first := chunk.ID(0).FirstLedger()
+	var contractID xdr.ContractId
+	contractID[0] = 0xab
+	events := func(n int) []xdr.ContractEvent {
+		evs := make([]xdr.ContractEvent, n)
+		for i := range evs {
+			evs[i] = rpcv2test.SymbolContractEvent(contractID, "x", "x")
+		}
+		return evs
+	}
+	// The first ledger fills index slab 0, whose seal fails; the second
+	// reaches the next slab boundary and learns of it.
+	_, err = svc.Ingest(first, xdr.LedgerCloseMetaView(rpcv2test.EventsLCMBytes(t, first, events(1<<16+1)...)))
+	require.NoError(t, err)
+	sink.hotPhases = nil
+	_, err = svc.Ingest(first+1, xdr.LedgerCloseMetaView(rpcv2test.EventsLCMBytes(t, first+1, events(1<<16)...)))
+	require.ErrorContains(t, err, "seal index slab 0")
+
+	require.Len(t, sink.hotPhases, int(hotchunk.NumPhases), "every phase emitted once")
+	items := sink.hotPhaseItems()
+	assert.Equal(t, 1, items[hotchunk.PhaseLedgers])
+	assert.Equal(t, 1, items[hotchunk.PhaseTxhash])
+	assert.Equal(t, 1<<16, items[hotchunk.PhaseEvents])
+	failed, hadErr := sink.hotPhaseErr()
+	require.True(t, hadErr)
+	assert.Equal(t, hotchunk.PhaseApply, failed)
+}
+
 // TestHotService_CommitErrorLandsOnCommitPhase asserts a commit failure (a closed
 // DB) surfaces the error on the commit phase — by construction, not by a
 // separately-maintained label — and emits no items on the failure path.

cmd/stellar-rpc/internal/rpcv2/ingest/metrics.go

@@ -60,11 +60,11 @@ var coldStagePairs = []struct{ dataType, stage string }{
 type MetricSink interface {
 	// HotPhase reports ONE phase of one hot ledger ingest — the single hot-tier
 	// signal family. It carries that phase's wall-clock, its item count (0 for the
-	// extract/commit/apply phases, the per-type write volume for the write phases, on
-	// the success path), and its outcome (err is non-nil only on the phase that
+	// extract/commit/apply phases, the per-type write volume for the write phases
+	// once the commit landed), and its outcome (err is non-nil only on the phase that
 	// failed, so a decode failure lands on PhaseExtract and a commit failure on
-	// PhaseCommit by construction; the post-commit PhaseApply runs on success only, so
-	// it never carries an error). The per-ledger total is the sum of the phase
+	// PhaseCommit by construction; the post-commit PhaseApply fails only when the
+	// events index could not seal a slab). The per-ledger total is the sum of the phase
 	// durations; the caller emits phases [0, Failed] on error and all phases on success.
 	HotPhase(phase hotchunk.Phase, d time.Duration, items int, err error)
 	// ColdIngest reports one cold ingester's per-chunk total: the summed Ingest

cmd/stellar-rpc/internal/rpcv2/ingest/service.go

@@ -41,11 +41,12 @@ func NewHotService(db *hotchunk.DB, windows *feewindow.FeeWindows, sink MetricSi
 // success it returns the committed ledger's close time (see
 // hotchunk.LedgerReport.CloseTime). Each
 // phase carries its own wall-clock (the phases partition the per-ledger total),
-// the write phases carry per-type item volume on success, and the outcome lands on
+// the write phases carry per-type item volume once the commit landed, and the outcome lands on
 // the phase that failed BY CONSTRUCTION — a decode failure on PhaseExtract, a
 // commit failure on PhaseCommit — so there is no mislabeled batch-scoped error.
-// On failure only phases [0, Failed] ran, so only those are emitted (and with zero
-// items — nothing landed durably); on success every phase is emitted. The walk
+// On failure only phases [0, Failed] ran, so only those are emitted; the write
+// phases carry their item volume once the commit landed, so a failure before
+// the commit carries none. On success every phase is emitted. The walk
 // runs here (see the type doc) and its duration folds into PhaseExtract, keeping
 // that phase "the walk + product reads".
 //
@@ -79,15 +80,11 @@ func (s *HotService) Ingest(seq uint32, lcmView xdr.LedgerCloseMetaView) (int64,
 		last = rep.Failed
 	}
 	for p := hotchunk.Phase(0); p <= last; p++ {
-		items := rep.Phases[p].Items
 		var perr error
-		if err != nil {
-			items = 0 // the failure path committed nothing durably
-			if p == rep.Failed {
-				perr = err
-			}
+		if err != nil && p == rep.Failed {
+			perr = err
 		}
-		s.sink.HotPhase(p, rep.Phases[p].Dur, items, perr)
+		s.sink.HotPhase(p, rep.Phases[p].Dur, rep.Phases[p].Items, perr)
 	}
 	if err != nil {
 		return 0, err

cmd/stellar-rpc/internal/rpcv2/stores/event/benchmark_test.go

@@ -7,11 +7,16 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	protocol "github.com/stellar/go-stellar-sdk/protocols/rpc"
 	"github.com/stellar/go-stellar-sdk/xdr"
+
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/rocksdb"
 )
 
-// BenchmarkEventIndex_10M measures heap at full chunk scale.
+// BenchmarkEventIndex_10M builds the hot index at full chunk scale and
+// measures the heap it holds once built.
 // Distribution modeled on real production chunk data:
 //
 //	chunk       events       terms     total_adds   mean_card   max_card
@@ -20,40 +25,48 @@ import (
 //	005908     9,255,090   2,289,828     37,397,684       16.3    6,440,193
 func BenchmarkEventIndex_10M(b *testing.B) {
 	for b.Loop() {
+		store, err := rocksdb.New(rocksdb.Config{
+			Path:           b.TempDir(),
+			ColumnFamilies: CFNames(),
+			Logger:         silentLogger(),
+			Tuning:         rocksdb.Tuning{BlockCacheMB: 64},
+			PerCFOptions:   CFOptions(),
+		})
+		require.NoError(b, err)
+		idx := newHotIndex(store, 0)
+
 		start := time.Now()
-		idx := buildIndex10M()
+		buildIndex10M(b, idx)
+		require.NoError(b, idx.settle())
 		buildSec := time.Since(start).Seconds()
 
 		b.ReportMetric(buildSec, "build_sec")
-		b.ReportMetric(float64(len(idx.terms)), "terms")
+		sealed, err := sealedHotSlabs(store)
+		require.NoError(b, err)
+		b.ReportMetric(float64(sealed), "sealed_slabs")
 
 		runtime.GC()
 		var mem runtime.MemStats
 		runtime.ReadMemStats(&mem)
 		b.ReportMetric(float64(mem.HeapInuse)/(1024*1024), "heap_MB")
 
 		runtime.KeepAlive(idx)
+		require.NoError(b, store.Close())
 	}
 }
 
 // buildIndex10M simulates a full chunk based on real production data:
 // ~9M events, ~2M unique terms, ~35M adds for the contract-ID and topic
 // terms, plus the 2 adds per event the type and topic-count families
-// contribute.
-//
-// Adds are grouped by term within a ledger, the way HotStore.applyLedger
-// batches them. Feeding a chunk-wide term one event at a time instead
-// pays ConcurrentBitmaps.AddTo's COW clone per event and costs several
-// times the build time, which no production path does.
-func buildIndex10M() *ConcurrentBitmaps {
+// contribute. It feeds the index one ledger at a time, as ingestion does.
+func buildIndex10M(b *testing.B, idx *hotIndex) {
 	const (
 		totalEvents     = 9_000_000
 		numContracts    = 10_000
 		numTopicVals    = 3_000_000
 		eventsPerLedger = 900
 	)
 
-	idx := NewConcurrentBitmapsFromBitmaps(NewBitmaps())
 	rng := rand.New(rand.NewSource(42))
 
 	contractKeys := make([]TermKey, numContracts)
@@ -77,34 +90,24 @@ func buildIndex10M() *ConcurrentBitmaps {
 	contractType := EventTypeTermKey(xdr.ContractEventTypeContract)
 	systemType := EventTypeTermKey(xdr.ContractEventTypeSystem)
 
-	perKeyIDs := make(map[TermKey][]uint32, 64)
-	flush := func() {
-		for key, ids := range perKeyIDs {
-			idx.AddTo(key, ids...)
-			delete(perKeyIDs, key)
-		}
-	}
-	add := func(key TermKey, eventID uint32) {
-		perKeyIDs[key] = append(perKeyIDs[key], eventID)
-	}
-
+	ledger := make([][]TermKey, 0, eventsPerLedger)
 	for eventID := range uint32(totalEvents) {
-		add(contractKeys[eventID%uint32(numContracts)], eventID)
+		keys := make([]TermKey, 0, maxTermsPerEvent)
+		keys = append(keys, contractKeys[eventID%uint32(numContracts)])
 		numTopics := rng.Intn(protocol.MaxTopicCount + 2)
-		for range numTopics {
-			add(topicKeys[zipf.Uint64()], eventID)
+		for range min(numTopics, protocol.MaxTopicCount) {
+			keys = append(keys, topicKeys[zipf.Uint64()])
 		}
 		if eventID%systemEventEvery == 0 {
-			add(systemType, eventID)
+			keys = append(keys, systemType)
 		} else {
-			add(contractType, eventID)
+			keys = append(keys, contractType)
 		}
-		add(TopicCountTermKey(numTopics), eventID)
-		if (eventID+1)%eventsPerLedger == 0 {
-			flush()
+		keys = append(keys, TopicCountTermKey(numTopics))
+		ledger = append(ledger, keys)
+		if len(ledger) == eventsPerLedger {
+			require.NoError(b, idx.add(eventID+1-eventsPerLedger, ledger))
+			ledger = ledger[:0]
 		}
 	}
-	flush()
-
-	return idx
 }

cmd/stellar-rpc/internal/rpcv2/stores/event/bitmaps.go

@@ -16,8 +16,6 @@ import (
 //
 // NOT safe for concurrent access. The caller guarantees serial
 // access (build then hand off, or a single goroutine throughout).
-// For the concurrent-reader-vs-single-writer case (live HotStore
-// ingest) use ConcurrentBitmaps.
 type Bitmaps map[TermKey]*roaring.Bitmap
 
 // NewBitmaps returns an empty Bitmaps. Equivalent to make(Bitmaps).