From d6c50bf0809f7f47d620b8f81ad5fd7b39fa5410 Mon Sep 17 00:00:00 2001 From: tamirms Date: Tue, 6 Oct 2026 00:45:32 +0100 Subject: [PATCH] events: seal the hot index one slab at a time The hot events index held every term of a chunk as an in-memory bitmap and wrote one events_index key per posting into each ledger's batch. Its memory grew with the number of distinct terms in the chunk, and the per-posting keys were most of each commit. Event ids are now cut into slabs of 65,536, the matcher's unit. The slab being filled is a fixed array of postings with per-bucket chains, which one writer appends to and readers walk without a lock. When the next slab starts, the full one is sealed in the background: its terms are written in key order as one file and loaded into events_index under slab || term. events_index has auto compaction off, and its files never overlap so compaction would have nothing to do; their index blocks live in the block cache. The writer waits for the previous seal before it starts the next one, so the index holds at most two slabs in memory whatever the chunk holds, and the ledger commit carries no index bytes any more. A failed seal keeps its slab readable from memory and fails the next ledger's apply, through the apply hook, which can now fail; the ledger report then names PhaseApply, and the write phases carry their item counts only once the commit landed. The next open indexes the unsealed events again from events_data, through the same path ingestion uses, and seals a full slab among them as soon as the next one starts. LookupKeys reads the slabs the window touches, from memory or from events_index, and returns a bitmap per key, empty rather than nil, since it has not seen the rest of the chunk. The bench sink counts an ingest total only for a ledger whose apply succeeded. The design documents describe the index as it is now. Measured with bench-ingest hot on the same synthetic ledgers (6,000 events of 7 terms each per ledger) at a 600 ms cadence, 600 ledgers, two runs each: per-ledger ingest p50 139 and 144 ms before, 70 and 71 ms after; p99 288 and 319 ms before, 82 and 88 ms after; the events, commit and apply phases went from 12, 72 and 11 ms to 4, 21 and 1 ms at p50; peak RSS 1,373 and 1,379 MB before, 549 and 545 MB after. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01BiCL19kUrno3Z4SAYgpAfh --- .../internal/rpcv2/bench/csvsink.go | 11 +- .../internal/rpcv2/bench/csvsink_test.go | 17 +- .../internal/rpcv2/ingest/ingest_test.go | 41 ++ .../internal/rpcv2/ingest/metrics.go | 8 +- .../internal/rpcv2/ingest/service.go | 17 +- .../rpcv2/stores/event/benchmark_test.go | 69 +-- .../internal/rpcv2/stores/event/bitmaps.go | 2 - .../rpcv2/stores/event/concurrent_bitmaps.go | 222 -------- .../stores/event/concurrent_bitmaps_test.go | 486 ----------------- .../event/concurrent_bitmaps_writer_test.go | 498 ------------------ .../internal/rpcv2/stores/event/hot_index.go | 318 +++++++++++ .../rpcv2/stores/event/hot_index_test.go | 339 ++++++++++++ .../internal/rpcv2/stores/event/hot_store.go | 311 +++++------ .../rpcv2/stores/event/hot_store_test.go | 140 ++--- .../rpcv2/stores/event/hot_warmup_test.go | 165 ++---- .../internal/rpcv2/stores/event/index_test.go | 2 +- .../internal/rpcv2/stores/event/match_test.go | 81 +-- .../stores/event/matches_differential_test.go | 109 ++-- .../internal/rpcv2/stores/event/reader.go | 24 +- .../stores/event/roaring_contract_test.go | 137 ++--- .../internal/rpcv2/stores/event/slab_match.go | 10 +- .../rpcv2/stores/event/slab_match_test.go | 12 +- .../rpcv2/stores/hotchunk/hotchunk.go | 48 +- .../rpcv2/stores/hotchunk/hotchunk_test.go | 72 ++- .../full-history-streaming-workflow.md | 8 +- design-docs/getevents-full-history-design.md | 55 +- 26 files changed, 1177 insertions(+), 2025 deletions(-) delete mode 100644 cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go delete mode 100644 cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_test.go delete mode 100644 cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_writer_test.go create mode 100644 cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index.go create mode 100644 cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index_test.go diff --git a/cmd/stellar-rpc/internal/rpcv2/bench/csvsink.go b/cmd/stellar-rpc/internal/rpcv2/bench/csvsink.go index 826bc81f7..7013fc06f 100644 --- a/cmd/stellar-rpc/internal/rpcv2/bench/csvsink.go +++ b/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) } } diff --git a/cmd/stellar-rpc/internal/rpcv2/bench/csvsink_test.go b/cmd/stellar-rpc/internal/rpcv2/bench/csvsink_test.go index b4fe1bd0b..5aac84a0c 100644 --- a/cmd/stellar-rpc/internal/rpcv2/bench/csvsink_test.go +++ b/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) diff --git a/cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go b/cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go index 394055028..19fba2f7e 100644 --- a/cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go +++ b/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. diff --git a/cmd/stellar-rpc/internal/rpcv2/ingest/metrics.go b/cmd/stellar-rpc/internal/rpcv2/ingest/metrics.go index d06bb5896..c4773ac7f 100644 --- a/cmd/stellar-rpc/internal/rpcv2/ingest/metrics.go +++ b/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 diff --git a/cmd/stellar-rpc/internal/rpcv2/ingest/service.go b/cmd/stellar-rpc/internal/rpcv2/ingest/service.go index fefd3f704..1a2ab23f4 100644 --- a/cmd/stellar-rpc/internal/rpcv2/ingest/service.go +++ b/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 diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/benchmark_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/benchmark_test.go index 414468e15..b2cf56fa7 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/benchmark_test.go +++ b/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,12 +25,25 @@ 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 @@ -33,19 +51,15 @@ func BenchmarkEventIndex_10M(b *testing.B) { 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 @@ -53,7 +67,6 @@ func buildIndex10M() *ConcurrentBitmaps { 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 } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/bitmaps.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/bitmaps.go index d8a988fd2..6e47248aa 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/bitmaps.go +++ b/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). diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go deleted file mode 100644 index 102e2f841..000000000 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go +++ /dev/null @@ -1,222 +0,0 @@ -package event - -import ( - "sync" - "sync/atomic" - - "github.com/RoaringBitmap/roaring/v2" -) - -// promotionThreshold is the number of event IDs a term holds in a -// sorted list before it is promoted to a roaring bitmap. Observed -// mean cardinality is ~14.5–16.3 events per term (production chunks -// 005901–005908; see BenchmarkEventIndex_10M), so most terms stay -// in list mode. -const promotionThreshold = 64 - -// denseState is the state of one dense term. One is allocated at -// promotion and lives for the term's lifetime. -// -// - wbm is the writer's private bitmap. AddTo mutates it in place. -// It is never handed to a reader. -// - pub is the last immutable snapshot handed to readers. nil means -// wbm holds writes that no snapshot has: nil at promotion, nil -// again after every write, non-nil only once a reader has cloned -// wbm since the last write. -// - mu serializes AddMany on wbm against Clone of wbm. Both write -// wbm's copy-on-write flags. Every store to pub happens under mu. -type denseState struct { - mu sync.Mutex - wbm *roaring.Bitmap - pub atomic.Pointer[roaring.Bitmap] -} - -// termState is the immutable per-term entry. Exactly one field is set. -// - Sparse: ids is the sorted event-ID list. AddTo publishes a new -// termState on every write. -// - Dense: dense is the term's denseState. Published once, at -// promotion; later snapshots go into dense.pub. -type termState struct { - ids []uint32 - dense *denseState -} - -// ConcurrentBitmaps is the in-memory event index for live ingest: -// one writer, many readers. -// -// rwmu protects only the terms map. Per-term state is reached through -// an atomic pointer. -// -// AddTo is single-writer: the orchestrator ingests from one goroutine -// per chunk. Get is safe to call concurrently with AddTo. Get takes -// the map's read lock for the lookup and no other lock, unless the -// term is dense and was written since its last snapshot; then one -// reader clones the writer's bitmap under the per-term mutex. AddTo -// on a dense term waits for a reader that is cloning that term. -type ConcurrentBitmaps struct { - rwmu sync.RWMutex - terms map[TermKey]*atomic.Pointer[termState] -} - -// NewConcurrentBitmapsFromBitmaps takes ownership of a Bitmaps built -// by warmup or backfill. The input must not be used afterwards. -// -// Terms below promotionThreshold become sparse lists, the same -// representation AddTo gives them. Terms at or above it keep their -// bitmap as the writer-private wbm. No snapshot is published here; -// the first Get on a dense term clones it. -func NewConcurrentBitmapsFromBitmaps(b Bitmaps) *ConcurrentBitmaps { - cb := &ConcurrentBitmaps{terms: make(map[TermKey]*atomic.Pointer[termState], len(b))} - for k, bm := range b { - if bm == nil { - continue - } - p := &atomic.Pointer[termState]{} - if bm.GetCardinality() < promotionThreshold { - p.Store(termStateFromIDs(bm.ToArray())) - } else { - p.Store(&termState{dense: newDenseState(bm)}) - } - cb.terms[k] = p - } - return cb -} - -// Get returns the bitmap for key, or (nil, nil) when key is not -// indexed. The result is read-only: dense terms share one bitmap -// across all concurrent readers, and the index never mutates it. -// -// Forbidden on the returned bitmap — these mutate internal state a -// concurrent reader or the writer may also be touching: -// - Clone, CloneCopyOnWriteContainers (COW is on: Clone writes its -// source's copy-on-write flags, so two readers cloning race) -// - RunOptimize, AddRange, RemoveRange, FlipInt -// - Add, AddMany, Remove, CheckedAdd, CheckedRemove, AddInt -// - SetCopyOnWrite -// - roaring.FastAnd / roaring.FastOr with a single input, which Clone it -// -// Safe: any non-mutating read (Contains, GetCardinality, Iterator, -// NextValue, PreviousValue, ToArray, IsEmpty, Minimum, Maximum) and -// passing it to FastAnd beside at least one other input; -// roaring_contract_test.go pins these against the pinned roaring version. -// -// A Get that starts after an AddTo returns sees that AddTo's IDs, and -// the pointer stays valid for as long as the caller holds it. The -// result is a point-in-time image: a sparse term is copied out of the -// published id list, and a dense term's snapshot is never mutated once -// published. Neither grows as ingest continues, so a query can hold it -// for a whole walk. -func (s *ConcurrentBitmaps) Get(key TermKey) (*roaring.Bitmap, error) { - s.rwmu.RLock() - p := s.terms[key] - s.rwmu.RUnlock() - if p == nil { - return nil, nil //nolint:nilnil // not-found is signaled by nil bitmap, no error - } - st := p.Load() - if st.dense != nil { - return st.dense.snapshot(), nil - } - bm := roaring.New() - bm.AddMany(st.ids) - return bm, nil -} - -// snapshot returns the term's current immutable bitmap. If a write -// landed since the last snapshot it clones wbm once and publishes -// the clone. -func (d *denseState) snapshot() *roaring.Bitmap { - if bm := d.pub.Load(); bm != nil { - return bm - } - d.mu.Lock() - defer d.mu.Unlock() - bm := d.pub.Load() - if bm == nil { - bm = d.wbm.Clone() - d.pub.Store(bm) - } - return bm -} - -// AddTo records each eventID under key. Callers feed events in -// event-ID order relative to the chunk, so a duplicate is a retry of an -// already-added prefix and is skipped. -func (s *ConcurrentBitmaps) AddTo(key TermKey, eventIDs ...uint32) { - if len(eventIDs) == 0 { - return - } - - s.rwmu.RLock() - p, ok := s.terms[key] - s.rwmu.RUnlock() - - if !ok { - next := newTermState(eventIDs) - p = &atomic.Pointer[termState]{} - p.Store(next) - s.rwmu.Lock() - s.terms[key] = p - s.rwmu.Unlock() - return - } - - old := p.Load() - if d := old.dense; d != nil { - d.mu.Lock() - defer d.mu.Unlock() - d.wbm.AddMany(eventIDs) - d.pub.Store(nil) - return - } - - // Sparse mode: new list, then promote if it crossed the threshold. - ids := make([]uint32, 0, len(old.ids)+len(eventIDs)) - ids = append(ids, old.ids...) - p.Store(termStateFromIDs(appendSorted(ids, eventIDs))) -} - -// appendSorted appends the ids in src that are greater than dst's -// last element. -func appendSorted(dst, src []uint32) []uint32 { - for _, id := range src { - if len(dst) > 0 && dst[len(dst)-1] >= id { - continue - } - dst = append(dst, id) - } - return dst -} - -// termStateFromIDs builds the termState for an ascending, unique id -// list, applying promotionThreshold. -func termStateFromIDs(ids []uint32) *termState { - if len(ids) >= promotionThreshold { - return promote(ids) - } - return &termState{ids: ids} -} - -// newDenseState wraps a writer-owned bitmap as a dense term. The -// bitmap is marked CopyOnWrite so snapshot's Clone is shallow. No -// snapshot is published; the first Get clones. -func newDenseState(bm *roaring.Bitmap) *denseState { - bm.SetCopyOnWrite(true) - return &denseState{wbm: bm} -} - -// promote builds the dense termState for a term from its sorted ids. -func promote(ids []uint32) *termState { - bm := roaring.New() - bm.AddMany(ids) - return &termState{dense: newDenseState(bm)} -} - -// newTermState builds the first termState for a new key. A batch at -// or above the threshold is promoted without building a list. -func newTermState(eventIDs []uint32) *termState { - if len(eventIDs) >= promotionThreshold { - return promote(eventIDs) - } - return termStateFromIDs(appendSorted(make([]uint32, 0, len(eventIDs)), eventIDs)) -} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_test.go deleted file mode 100644 index 2234c5d33..000000000 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_test.go +++ /dev/null @@ -1,486 +0,0 @@ -package event - -import ( - "sync" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// newTestConcurrentBitmaps builds an empty ConcurrentBitmaps via the -// only remaining constructor (production always converts from a -// warmup/backfill-built Bitmaps). -func newTestConcurrentBitmaps() *ConcurrentBitmaps { - return NewConcurrentBitmapsFromBitmaps(NewBitmaps()) -} - -func TestConcurrentBitmaps_AddToAndGet(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("transfer"), FieldTopic0) - - s.AddTo(key, 0) - s.AddTo(key, 1) - s.AddTo(key, 2) - - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(3), bm.GetCardinality()) - assert.True(t, bm.Contains(0)) - assert.True(t, bm.Contains(1)) - assert.True(t, bm.Contains(2)) -} - -func TestConcurrentBitmaps_GetMissing(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("missing"), FieldTopic0) - bm, err := s.Get(key) - require.NoError(t, err) - assert.Nil(t, bm) -} - -func TestConcurrentBitmaps_ListMode(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("sparse"), FieldTopic0) - - for i := range uint32(promotionThreshold - 1) { - s.AddTo(key, i) - } - - p := s.terms[key] - require.NotNil(t, p) - st := p.Load() - require.NotNil(t, st) - assert.Nil(t, st.dense, "must still be in list mode") - assert.Len(t, st.ids, promotionThreshold-1) - - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(promotionThreshold-1), bm.GetCardinality()) - - // Get must not have promoted. - assert.Nil(t, p.Load().dense) -} - -func TestConcurrentBitmaps_Promotion(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("dense"), FieldTopic0) - - for i := range uint32(promotionThreshold) { - s.AddTo(key, i) - } - - p := s.terms[key] - require.NotNil(t, p) - st := p.Load() - require.NotNil(t, st.dense) - assert.Empty(t, st.ids, "sparse ids cleared after promotion") - bm, err := s.Get(key) - require.NoError(t, err) - assert.Equal(t, uint64(promotionThreshold), bm.GetCardinality()) -} - -func TestConcurrentBitmaps_AddAfterPromotion(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("dense"), FieldTopic0) - - for i := range uint32(promotionThreshold) { - s.AddTo(key, i) - } - s.AddTo(key, 1000) - s.AddTo(key, 2000) - - bm, err := s.Get(key) - require.NoError(t, err) - assert.Equal(t, uint64(promotionThreshold+2), bm.GetCardinality()) - assert.True(t, bm.Contains(1000)) - assert.True(t, bm.Contains(2000)) -} - -func TestConcurrentBitmaps_BatchAddTo(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("batch"), FieldTopic0) - - s.AddTo(key, 0, 1, 2, 3, 4) - - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(5), bm.GetCardinality()) - assert.True(t, bm.Contains(0)) - assert.True(t, bm.Contains(4)) -} - -func TestConcurrentBitmaps_BatchAddToPromotion(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("batch-promote"), FieldTopic0) - - // Single batch call that crosses threshold. - ids := make([]uint32, promotionThreshold+10) - for i := range ids { - ids[i] = uint32(i) - } - s.AddTo(key, ids...) - - p := s.terms[key] - require.NotNil(t, p) - st := p.Load() - require.NotNil(t, st.dense, "single-batch over threshold must promote immediately") - bm, err := s.Get(key) - require.NoError(t, err) - assert.Equal(t, uint64(promotionThreshold+10), bm.GetCardinality()) -} - -// TestConcurrentBitmaps_GetReturnsImmutableSnapshot pins the -// COW-on-write contract: a bitmap returned by Get is an immutable -// snapshot, so a subsequent AddTo (which produces a new snapshot -// via atomic.Store) does NOT mutate the previously-returned -// pointer. This is the key invariant readers can rely on across the -// borrow. -func TestConcurrentBitmaps_GetReturnsImmutableSnapshot(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("borrow"), FieldTopic0) - - // Promote to bitmap mode. - for i := range uint32(promotionThreshold) { - s.AddTo(key, i) - } - - before, err := s.Get(key) - require.NoError(t, err) - beforeCard := before.GetCardinality() - - // New AddTo publishes a new snapshot via atomic.Store. - s.AddTo(key, 9_999_999) - - // before still observes the pre-AddTo cardinality. - assert.Equal(t, beforeCard, before.GetCardinality(), - "AddTo published a new snapshot; the borrowed pointer must remain unchanged") - - after, err := s.Get(key) - require.NoError(t, err) - assert.True(t, after.Contains(9_999_999), - "subsequent Get must observe the new snapshot") -} - -// TestConcurrentBitmaps_ConcurrentGetIsSafe runs many concurrent -// Get callers against the same store. Get is lock-free past the -// brief map-lookup RLock and returns an immutable snapshot, so -// concurrent reads should not race. Run under -race. -func TestConcurrentBitmaps_ConcurrentGetIsSafe(t *testing.T) { - s := newTestConcurrentBitmaps() - const nTerms = 200 - keys := make([]TermKey, nTerms) - for i := range nTerms { - k := ComputeTermKey([]byte{byte(i / 256), byte(i % 256)}, FieldTopic0) - keys[i] = k - idCount := uint32(promotionThreshold + 1) - if i%2 == 0 { - idCount = 3 - } - ids := make([]uint32, idCount) - for j := range ids { - ids[j] = uint32(j) - } - s.AddTo(k, ids...) - } - - const numReaders = 8 - var wg sync.WaitGroup - for range numReaders { - wg.Go(func() { - for range 50 { - for _, k := range keys { - bm, err := s.Get(k) - require.NoError(t, err) - require.NotNil(t, bm) - _ = bm.Contains(0) - } - } - }) - } - wg.Wait() - - // Sanity check: the store is untouched by all that reading. - for _, k := range keys { - bm, err := s.Get(k) - require.NoError(t, err) - require.NotNil(t, bm) - } -} - -// TestConcurrentBitmaps_ConcurrentReadWrite exercises the COW -// contract under a single writer and many readers. Readers atomic- -// Load the current snapshot and operate on it independently while -// the writer publishes new snapshots; no clones or locks span the -// borrow. Under -race no data races should be reported. -func TestConcurrentBitmaps_ConcurrentReadWrite(t *testing.T) { - s := newTestConcurrentBitmaps() - - const numTerms = 100 - const numEvents = 10_000 - const numReaders = 4 - - keys := make([]TermKey, numTerms) - for i := range keys { - keys[i] = ComputeTermKey([]byte{byte(i)}, FieldTopic0) - } - - var wg sync.WaitGroup - - wg.Go(func() { - for i := range uint32(numEvents) { - s.AddTo(keys[i%numTerms], i) - } - }) - - for range numReaders { - wg.Go(func() { - for i := range numEvents { - _, _ = s.Get(keys[i%numTerms]) - } - }) - } - - wg.Wait() - - for _, key := range keys { - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(numEvents/numTerms), bm.GetCardinality()) - } -} - -// TestConcurrentBitmaps_GetDuringPromotionNeverReturnsNil pins -// concurrent-reader safety across the sparse→dense promotion -// transition. The current termState design publishes the whole -// (ids, bm) pair via a single atomic.Store, so the -// observability bug it was originally added to catch (a reader's -// two Loads of separate ids/bm atomic.Pointers straddling two -// separate Stores and seeing (bm=nil, ids=empty)) is structurally -// impossible. The test still has value as a -race probe: many -// readers calling Get while a writer drives terms across the -// promotion boundary should never produce a nil bitmap and -// should never trip the race detector. -func TestConcurrentBitmaps_GetDuringPromotionNeverReturnsNil(t *testing.T) { - s := newTestConcurrentBitmaps() - const numKeys = 200 - - keys := make([]TermKey, numKeys) - for i := range numKeys { - keys[i] = ComputeTermKey([]byte{byte(i / 256), byte(i % 256)}, FieldTopic0) - } - - // Seed each term with promotionThreshold-1 ids: sparse mode, - // one event away from promotion. - for _, k := range keys { - ids := make([]uint32, promotionThreshold-1) - for j := range ids { - ids[j] = uint32(j) - } - s.AddTo(k, ids...) - } - - var wg sync.WaitGroup - const numReaders = 8 - - stop := make(chan struct{}) - - // Writer goroutine: trigger promotion on each key by appending - // one more event each cycle. After all keys promote it stops. - wg.Go(func() { - defer close(stop) - for i, k := range keys { - s.AddTo(k, uint32(promotionThreshold-1+i)) - } - }) - - for range numReaders { - wg.Go(func() { - for { - select { - case <-stop: - return - default: - } - for _, k := range keys { - bm, err := s.Get(k) - require.NoError(t, err) - // The term was seeded with promotionThreshold-1 - // ids and the writer only appends — Get must - // always observe a non-nil bitmap. - require.NotNil(t, bm, "Get returned nil during promotion window") - } - } - }) - } - wg.Wait() -} - -// TestConcurrentBitmaps_AddToIsIdempotent pins the dedup contract: -// AddTo can be called multiple times with the same eventID for the -// same key and the result is the same as adding it once. Covers -// both list mode (sorted-prefix check) and bitmap mode (roaring's -// set semantics). -func TestConcurrentBitmaps_AddToIsIdempotent(t *testing.T) { - t.Run("list mode", func(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("sparse"), FieldTopic0) - - // Add a few in order. - s.AddTo(key, 0) - s.AddTo(key, 1) - s.AddTo(key, 2) - - // Replay (simulates a phase-3 retry after partial failure). - s.AddTo(key, 0) - s.AddTo(key, 1) - s.AddTo(key, 2) - // Also replay multiple at once. - s.AddTo(key, 1, 2) - // And add a new one — must still go through. - s.AddTo(key, 3) - - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(4), bm.GetCardinality()) - for _, id := range []uint32{0, 1, 2, 3} { - assert.True(t, bm.Contains(id)) - } - }) - - t.Run("bitmap mode", func(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("dense"), FieldTopic0) - - // Force bitmap mode by exceeding the threshold. - for i := range uint32(promotionThreshold) { - s.AddTo(key, i) - } - p := s.terms[key] - require.NotNil(t, p.Load().dense, "must have promoted to bitmap mode") - - // Replay — bitmap.AddMany is set-semantic, so no cardinality change. - for i := range uint32(promotionThreshold) { - s.AddTo(key, i) - } - - bm, err := s.Get(key) - require.NoError(t, err) - assert.Equal(t, uint64(promotionThreshold), bm.GetCardinality()) - }) -} - -// assertDenseCOW checks that a term is dense and that CopyOnWrite is -// enabled on both halves of its state: the writer-private bitmap -// (where it makes the reader-side Clone shallow) and the snapshot -// handed to readers (which inherits the flag through that Clone). -func assertDenseCOW(t *testing.T, s *ConcurrentBitmaps, key TermKey, msg string) { - t.Helper() - p := s.terms[key] - require.NotNil(t, p) - d := p.Load().dense - require.NotNil(t, d, "term must be dense") - assert.True(t, d.wbm.GetCopyOnWrite(), msg+" (writer bitmap)") - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.True(t, bm.GetCopyOnWrite(), msg+" (published snapshot)") -} - -// TestConcurrentBitmaps_DenseAddToSetsCopyOnWrite pins that the -// dense path on AddTo (both the promotion transition in AddTo -// itself and newTermState's over-threshold initial batch) sets -// CopyOnWrite. The perf design relies on this: a regression that -// drops SetCopyOnWrite would silently make every reader-side Clone -// deep-copy the whole bitmap -// (+40% hot-ingest wall, observed empirically). -func TestConcurrentBitmaps_DenseAddToSetsCopyOnWrite(t *testing.T) { - t.Run("via promotion in AddTo", func(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("promote"), FieldTopic0) - for i := range uint32(promotionThreshold) { - s.AddTo(key, i) - } - assertDenseCOW(t, s, key, - "dense bitmap after promotion must have CopyOnWrite enabled") - }) - - t.Run("via newTermState over-threshold initial batch", func(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("initial"), FieldTopic0) - ids := make([]uint32, promotionThreshold+10) - for i := range ids { - ids[i] = uint32(i) - } - s.AddTo(key, ids...) - assertDenseCOW(t, s, key, - "dense bitmap from over-threshold initial AddTo must have CopyOnWrite enabled") - }) - - t.Run("subsequent AddTo preserves CopyOnWrite via Clone", func(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("evolve"), FieldTopic0) - for i := range uint32(promotionThreshold) { - s.AddTo(key, i) - } - s.AddTo(key, 10_000, 10_001, 10_002) - assertDenseCOW(t, s, key, - "dense bitmap after additional AddTos must keep CopyOnWrite (inherited via Clone)") - }) -} - -// TestNewConcurrentBitmapsFromBitmaps_DirectlyPinsContract verifies -// the warmup-side conversion constructor: -// - input bitmaps survive in the result with their cardinality -// intact; -// - terms below promotionThreshold become sparse lists, terms at or -// above it become dense and keep CopyOnWrite; -// - nil bitmaps in the input map are skipped; -// - a subsequent AddTo on a converted dense term is visible to the -// next Get. -func TestNewConcurrentBitmapsFromBitmaps_DirectlyPinsContract(t *testing.T) { - src := NewBitmaps() - keySparse := ComputeTermKey([]byte("a"), FieldTopic0) - keyDense := ComputeTermKey([]byte("b"), FieldTopic1) - keyNil := ComputeTermKey([]byte("nil"), FieldTopic2) - - src.AddTo(keySparse, 0, 1, 2, 3, 4) - dense := make([]uint32, promotionThreshold) - for i := range dense { - dense[i] = uint32(i) * 1_000 - } - src.AddTo(keyDense, dense...) - src[keyNil] = nil - - cb := NewConcurrentBitmapsFromBitmaps(src) - - bmSparse, err := cb.Get(keySparse) - require.NoError(t, err) - require.NotNil(t, bmSparse) - assert.Equal(t, uint64(5), bmSparse.GetCardinality()) - assert.Nil(t, cb.terms[keySparse].Load().dense, "a sub-threshold warmup term must be sparse") - - bmDense, err := cb.Get(keyDense) - require.NoError(t, err) - require.NotNil(t, bmDense) - assert.Equal(t, uint64(len(dense)), bmDense.GetCardinality()) - assert.True(t, bmDense.GetCopyOnWrite()) - assert.NotNil(t, cb.terms[keyDense].Load().dense, "a warmup term at the threshold must be dense") - - bmNil, err := cb.Get(keyNil) - require.NoError(t, err) - assert.Nil(t, bmNil, "nil source entries must be skipped, not panicked") - - cb.AddTo(keyDense, 999_999) - post, err := cb.Get(keyDense) - require.NoError(t, err) - require.NotNil(t, post) - assert.True(t, post.GetCopyOnWrite()) - assert.Equal(t, uint64(len(dense)+1), post.GetCardinality()) - assert.Equal(t, uint64(len(dense)), bmDense.GetCardinality(), "earlier snapshot stays immutable") -} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_writer_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_writer_test.go deleted file mode 100644 index cfd493caa..000000000 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_writer_test.go +++ /dev/null @@ -1,498 +0,0 @@ -package event - -import ( - "runtime" - "sync" - "sync/atomic" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// TestConcurrentBitmaps_SnapshotSurvivesMultiContainerWrites: a -// snapshot handed to a reader is never mutated afterwards. The ids -// span ~16 containers so the writer both deep-copies shared -// containers and inserts new ones. -func TestConcurrentBitmaps_SnapshotSurvivesMultiContainerWrites(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("multi-container"), FieldTopic0) - - // Promote and seed one container. - seed := make([]uint32, promotionThreshold) - for i := range seed { - seed[i] = uint32(i) - } - s.AddTo(key, seed...) - - held, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, held) - heldCard := held.GetCardinality() - require.Equal(t, uint64(promotionThreshold), heldCard) - heldCopy := held.ToArray() - - // 1_000 more ids spread over 16 fresh containers, written one - // small batch per "ledger" so the writer touches an already-shared - // container repeatedly. - added := make([]uint32, 0, 100*10) - for batch := range 100 { - ids := make([]uint32, 10) - for j := range ids { - ids[j] = uint32(batch%16)*65_536 + 1_000 + uint32(batch)*10 + uint32(j) - } - s.AddTo(key, ids...) - added = append(added, ids...) - } - - assert.Equal(t, heldCard, held.GetCardinality(), - "a snapshot returned by Get must not be mutated by later AddTo calls") - assert.Equal(t, heldCopy, held.ToArray(), - "a snapshot returned by Get must keep exactly its original ids") - for _, id := range added { - assert.False(t, held.Contains(id), - "the held snapshot must not observe ids added after it was taken") - } - - fresh, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, fresh) - assert.Equal(t, heldCard+uint64(len(added)), fresh.GetCardinality(), - "a Get after AddTo must observe every write") - for _, id := range added { - require.True(t, fresh.Contains(id), "fresh snapshot missing id %d", id) - } -} - -// TestConcurrentBitmaps_GetAfterAddToIsFresh pins the freshness -// contract call by call: every single AddTo must be visible to the -// very next Get. The published snapshot is allowed to lag, but only -// until someone reads. -func TestConcurrentBitmaps_GetAfterAddToIsFresh(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("freshness"), FieldTopic0) - - for i := range uint32(promotionThreshold + 500) { - s.AddTo(key, i) - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - require.Equal(t, uint64(i+1), bm.GetCardinality(), - "Get after AddTo(%d) must see every id added so far", i) - require.True(t, bm.Contains(i)) - } -} - -// TestConcurrentBitmaps_WriterReaderRace runs one writer against -// eight readers that both snapshot and read through the snapshot -// (Contains / GetCardinality / Iterator). Under -race this pins that -// the writer's in-place AddMany and the readers' Clone never overlap -// on the writer-private bitmap, and that nothing mutates a snapshot -// while another goroutine reads it. The writer starts only once every -// reader is in its loop, and each reader keeps going for minReads -// iterations past the writer's end, so the test cannot pass without -// reader work overlapping writes. -func TestConcurrentBitmaps_WriterReaderRace(t *testing.T) { - s := newTestConcurrentBitmaps() - - const numTerms = 32 - const numLedgers = 400 - const perLedger = 8 - const numReaders = 8 - const minReads = 64 - - keys := make([]TermKey, numTerms) - for i := range keys { - keys[i] = ComputeTermKey([]byte{0xAA, byte(i)}, FieldTopic1) - } - - var done atomic.Bool - var reads atomic.Uint64 - var ready, wg sync.WaitGroup - ready.Add(numReaders) - - wg.Go(func() { - defer done.Store(true) - ready.Wait() - var next uint32 - for range numLedgers { - for _, key := range keys { - ids := make([]uint32, perLedger) - for j := range ids { - ids[j] = next - next++ - } - s.AddTo(key, ids...) - } - } - }) - - for r := range numReaders { - wg.Go(func() { - ready.Done() - for i, n := 0, 0; !done.Load() || n < minReads; i++ { - key := keys[(i+r)%numTerms] - bm, err := s.Get(key) - if err != nil || bm == nil { - continue - } - n++ - reads.Add(1) - _ = bm.GetCardinality() - _ = bm.Contains(uint32(i)) - it := bm.Iterator() - for n := 0; n < 64 && it.HasNext(); n++ { - _ = it.Next() - } - } - }) - } - - wg.Wait() - assert.GreaterOrEqual(t, reads.Load(), uint64(numReaders*minReads), - "every reader must have done real reads") - - want := uint64(numLedgers * perLedger) - for _, key := range keys { - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, want, bm.GetCardinality()) - } -} - -// TestConcurrentBitmaps_PromotionMidStreamUnderReaders drives the -// sparse→dense transition while readers are live, then keeps writing -// past it. Readers must never see a nil bitmap, never see a -// cardinality that goes backwards, and must see the full set at the -// end. -func TestConcurrentBitmaps_PromotionMidStreamUnderReaders(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("promote-midstream"), FieldTopic2) - - const total = promotionThreshold * 8 - const numReaders = 4 - const minReads = 64 - - var writes atomic.Uint32 - var done atomic.Bool - var reads atomic.Uint64 - var ready, wg sync.WaitGroup - ready.Add(numReaders) - - wg.Go(func() { - defer done.Store(true) - ready.Wait() - for i := range uint32(total) { - s.AddTo(key, i) - writes.Store(i + 1) - } - }) - - for range numReaders { - wg.Go(func() { - ready.Done() - var last uint64 - for n := 0; !done.Load() || n < minReads; { - bm, err := s.Get(key) - if err != nil || bm == nil { - continue - } - n++ - reads.Add(1) - card := bm.GetCardinality() - assert.GreaterOrEqual(t, card, last, - "a term's observed cardinality must never go backwards") - assert.LessOrEqual(t, card, uint64(writes.Load())+1, - "a reader must not observe more ids than were written") - last = card - } - }) - } - - wg.Wait() - assert.GreaterOrEqual(t, reads.Load(), uint64(numReaders*minReads), - "every reader must have done real reads") - - bm, err := s.Get(key) - require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(total), bm.GetCardinality()) -} - -// TestConcurrentBitmaps_WarmupThenAddToThenGet covers the warmup -// entry path: NewConcurrentBitmapsFromBitmaps takes ownership of the -// built bitmaps without publishing a snapshot, so the first Get has -// to materialize one. A write before that first read must still be -// visible. -func TestConcurrentBitmaps_WarmupThenAddToThenGet(t *testing.T) { - src := NewBitmaps() - keyRead := ComputeTermKey([]byte("warm-read-first"), FieldTopic0) - keyWrite := ComputeTermKey([]byte("warm-write-first"), FieldTopic0) - - seed := make([]uint32, 200) - for i := range seed { - seed[i] = uint32(i) * 1_000 // spans several containers - } - src.AddTo(keyRead, seed...) - src.AddTo(keyWrite, seed...) - - cb := NewConcurrentBitmapsFromBitmaps(src) - - // Read first, then write, then read again. - first, err := cb.Get(keyRead) - require.NoError(t, err) - require.NotNil(t, first) - require.Equal(t, uint64(len(seed)), first.GetCardinality()) - cb.AddTo(keyRead, 999_999) - assert.Equal(t, uint64(len(seed)), first.GetCardinality(), - "the warmup snapshot handed out earlier must stay immutable") - second, err := cb.Get(keyRead) - require.NoError(t, err) - assert.Equal(t, uint64(len(seed)+1), second.GetCardinality()) - assert.True(t, second.Contains(999_999)) - - // Write before the first read ever happens. - cb.AddTo(keyWrite, 888_888) - only, err := cb.Get(keyWrite) - require.NoError(t, err) - require.NotNil(t, only) - assert.Equal(t, uint64(len(seed)+1), only.GetCardinality()) - assert.True(t, only.Contains(888_888)) - assert.True(t, only.GetCopyOnWrite(), - "warmup snapshots must keep CopyOnWrite so republishing stays shallow") -} - -// TestConcurrentBitmaps_UnreadTermNeverClones pins the allocation -// property the fix exists for: a dense term that nobody reads -// publishes no snapshot at all, so neither the entry's termState nor -// the term's published bitmap moves across thousands of AddTo calls. -// A regression that re-published per AddTo would move one of them. -func TestConcurrentBitmaps_UnreadTermNeverClones(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("unread"), FieldTopic0) - - seed := make([]uint32, promotionThreshold) - for i := range seed { - seed[i] = uint32(i) - } - s.AddTo(key, seed...) - - entry := s.terms[key].Load() - d := entry.dense - require.NotNil(t, d) - require.Nil(t, d.pub.Load(), "an unread dense term must publish nothing at promotion") - - for i := range uint32(5_000) { - s.AddTo(key, 100_000+i) - } - assert.Same(t, entry, s.terms[key].Load(), - "AddTo on a dense term must not publish a new termState") - assert.Nil(t, d.pub.Load(), - "AddTo on a dense term must not publish a snapshot") - - bm, err := s.Get(key) - require.NoError(t, err) - assert.Equal(t, uint64(promotionThreshold+5_000), bm.GetCardinality()) - assert.Same(t, entry, s.terms[key].Load(), - "a dense term's termState is published once and never replaced") - assert.Same(t, bm, d.pub.Load(), - "the first Get after writes must publish the snapshot it returns") - - // A second Get with no intervening write returns the same pointer. - again, err := s.Get(key) - require.NoError(t, err) - assert.Same(t, bm, again, - "Get must return the same pointer when nothing changed") -} - -// TestConcurrentBitmaps_DenseStateAfterAddToAndGet checks the state -// of a dense term at each step: no snapshot after promotion and after -// a write, the returned snapshot in pub after a read. Earlier -// snapshots stay frozen. -func TestConcurrentBitmaps_DenseStateAfterAddToAndGet(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("publish-order"), FieldTopic0) - - seed := make([]uint32, promotionThreshold) - for i := range seed { - seed[i] = uint32(i) - } - s.AddTo(key, seed...) - - d := s.terms[key].Load().dense - require.NotNil(t, d) - assert.Nil(t, d.pub.Load(), "a freshly promoted term has no snapshot until first read") - - first, err := s.Get(key) - require.NoError(t, err) - assert.Same(t, first, d.pub.Load(), - "the publishing reader stores the snapshot it returns") - - s.AddTo(key, 12_345) - assert.Nil(t, d.pub.Load(), "AddTo clears the stale snapshot before returning") - assert.False(t, first.Contains(12_345), "an earlier snapshot stays frozen") - - second, err := s.Get(key) - require.NoError(t, err) - assert.True(t, second.Contains(12_345), "the next Get observes the write") - assert.Same(t, second, d.pub.Load()) - assert.False(t, first.Contains(12_345), "the earlier snapshot is still frozen") -} - -// freshnessWriterCounters is the shared state between the freshness -// stress writer and its readers. -type freshnessWriterCounters struct { - committed *atomic.Uint32 // highest ID whose AddTo has returned - observed *atomic.Uint32 // highest committed ID a reader has seen in a snapshot - batches *atomic.Uint32 // batches the writer has finished -} - -// freshnessWriter adds numBatches three-ID batches under key, handing -// off to the readers after each one: it does not start the next batch -// until some reader has returned a snapshot holding this batch's last -// ID. That snapshot must be a clone made after the AddTo, so every -// batch forces at least one republish, and the readers race each -// other to make it. Returns early once the test has failed so a -// reader that gave up cannot hang the writer. -func freshnessWriter( - t *testing.T, s *ConcurrentBitmaps, key TermKey, c freshnessWriterCounters, - numBatches, firstID, idStride uint32, -) { - next := firstID - for i := range numBatches { - batch := []uint32{next, next + 977, next + 65_536} - s.AddTo(key, batch...) - last := batch[len(batch)-1] - c.committed.Store(last) - c.batches.Store(i + 1) - next += idStride - for c.observed.Load() < last { - if t.Failed() { - return - } - runtime.Gosched() - } - } -} - -// storeMax raises v to want unless v is already at least want. -func storeMax(v *atomic.Uint32, want uint32) { - for { - seen := v.Load() - if seen >= want || v.CompareAndSwap(seen, want) { - return - } - } -} - -// TestConcurrentBitmaps_FreshnessUnderConcurrentPublishers: the -// writer publishes the highest ID it has finished adding; each reader -// loads that ID, calls Get, and asserts the snapshot contains it. The -// batch count is fixed so the ids stay below MaxUint32 and never -// repeat an id an older snapshot already holds. -func TestConcurrentBitmaps_FreshnessUnderConcurrentPublishers(t *testing.T) { - s := newTestConcurrentBitmaps() - key := ComputeTermKey([]byte("freshness-stress"), FieldTopic0) - - // Seed ~200 containers so a republish Clone is long enough for a - // concurrent reader to interleave with it. - seed := make([]uint32, 0, 200*8) - for c := range uint32(200) { - for j := range uint32(8) { - seed = append(seed, c*65_536+j) - } - } - s.AddTo(key, seed...) - - numReaders := max(16, 2*runtime.GOMAXPROCS(0)) - // firstID + numBatches*idStride + 65_536 stays below MaxUint32. - const ( - numBatches = 1_000 - firstID = uint32(20_000_000) - idStride = uint32(131_072) - ) - - var committed, observed, batches atomic.Uint32 - var done atomic.Bool - var reads atomic.Uint64 - var wg sync.WaitGroup - - wg.Go(func() { - defer done.Store(true) - freshnessWriter(t, s, key, freshnessWriterCounters{ - committed: &committed, observed: &observed, batches: &batches, - }, numBatches, firstID, idStride) - }) - - for range numReaders { - wg.Go(func() { - for !done.Load() { - want := committed.Load() - if want == 0 { - runtime.Gosched() - continue - } - bm, err := s.Get(key) - if err != nil || bm == nil { - continue - } - reads.Add(1) - // Yield: every read after a write takes the term - // mutex, and unyielding readers starve the writer. - runtime.Gosched() - // Get started after the AddTo that committed `want` - // returned, so the snapshot must contain it. - if !bm.Contains(want) { - t.Errorf("Get returned a snapshot missing id %d, "+ - "committed before this Get started (cardinality %d)", - want, bm.GetCardinality()) - return - } - storeMax(&observed, want) - } - }) - } - - wg.Wait() - t.Logf("freshness stress: %d reads, %d batches", - reads.Load(), batches.Load()) - assert.Equal(t, uint32(numBatches), batches.Load(), "the writer must finish every batch") - assert.GreaterOrEqual(t, observed.Load(), committed.Load(), - "a reader must observe the final committed batch") - require.Positive(t, reads.Load(), "the stress loop must have done real reads") -} - -// TestConcurrentBitmaps_WarmupSubThresholdTermStaysSparse pins the -// warmup representation of a small term: it is a sparse list, Get -// returns its ids, later AddTo calls stay in sparse mode, and the term -// promotes at the threshold like a term built by AddTo. -func TestConcurrentBitmaps_WarmupSubThresholdTermStaysSparse(t *testing.T) { - src := NewBitmaps() - key := ComputeTermKey([]byte("warm-small"), FieldTopic0) - seed := make([]uint32, promotionThreshold-2) - for i := range seed { - seed[i] = uint32(i) * 7 - } - src.AddTo(key, seed...) - - cb := NewConcurrentBitmapsFromBitmaps(src) - st := cb.terms[key].Load() - require.Nil(t, st.dense) - assert.Equal(t, seed, st.ids) - - bm, err := cb.Get(key) - require.NoError(t, err) - assert.Equal(t, uint64(len(seed)), bm.GetCardinality()) - - next := seed[len(seed)-1] + 1 - cb.AddTo(key, next) - require.Nil(t, cb.terms[key].Load().dense, "one below the threshold stays sparse") - - cb.AddTo(key, next+1) - require.NotNil(t, cb.terms[key].Load().dense, "reaching the threshold promotes") - bm, err = cb.Get(key) - require.NoError(t, err) - assert.Equal(t, uint64(len(seed)+2), bm.GetCardinality()) - assert.True(t, bm.Contains(next+1)) -} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index.go new file mode 100644 index 000000000..8db1ed71e --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index.go @@ -0,0 +1,318 @@ +package event + +import ( + "bytes" + "cmp" + "context" + "encoding/binary" + "fmt" + "iter" + "math" + "slices" + "sync/atomic" + + "github.com/RoaringBitmap/roaring/v2" + + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/rocksdb" +) + +const ( + // hotSlabEvents is how many event ids a slab spans. + hotSlabEvents = 1 << indexSlabShift + + // hotSlabPostings is the most postings a slab holds: every one of its + // events carrying maxTermsPerEvent terms. + hotSlabPostings = hotSlabEvents * maxTermsPerEvent + + // hotIndexKeyLen is an IndexCF key: slab (2 BE) || term. + hotIndexKeyLen = 2 + len(TermKey{}) +) + +func hotIndexKey(slab uint32, k TermKey) []byte { + key := make([]byte, hotIndexKeyLen) + binary.BigEndian.PutUint16(key, uint16(slab)) //nolint:gosec // a chunk has at most 1<<16 slabs + copy(key[2:], k[:]) + return key +} + +// A posting's id within its slab, the slab number in an IndexCF key and a +// hash bucket are 16 bits each. +const ( + _ uint = indexSlabShift - 16 + _ uint = 16 - indexSlabShift +) + +type hotPosting struct { + key TermKey + next uint32 // 1 + position of the bucket's previous posting; 0 ends the chain + id uint16 // the event id within the slab +} + +// hotSlab holds the postings of one slab of event ids in arrival order. The +// postings of a hash bucket form a chain, newest first. One goroutine +// appends; any number read while it does. +type hotSlab struct { + slab uint32 + n uint32 + posts []hotPosting + // heads holds, per bucket, 1 + the position of its newest posting, or 0. + // A bucket is a key's leading 16 bits, so bucket order is key order. + heads []atomic.Uint32 +} + +func newHotSlab(slab uint32) *hotSlab { + return &hotSlab{ + slab: slab, + posts: make([]hotPosting, hotSlabPostings), + heads: make([]atomic.Uint32, hotSlabEvents), + } +} + +func (s *hotSlab) head(k TermKey) *atomic.Uint32 { + return &s.heads[binary.BigEndian.Uint16(k[:])] +} + +// add appends one posting. The head is stored last: a reader that loads it +// finds the posting, and everything it links to, already written. +func (s *hotSlab) add(k TermKey, id uint32) { + head := s.head(k) + s.posts[s.n] = hotPosting{key: k, next: head.Load(), id: uint16(id)} //nolint:gosec // the id within the slab + s.n++ + head.Store(s.n) +} + +// bitmap returns the term's event ids in this slab, nil if it has none. +func (s *hotSlab) bitmap(k TermKey) *roaring.Bitmap { + var bits [hotSlabEvents / 64]uint64 + found := false + for p := s.head(k).Load(); p != 0; p = s.posts[p-1].next { + if e := &s.posts[p-1]; e.key == k { + bits[e.id>>6] |= 1 << (e.id & 63) + found = true + } + } + if !found { + return nil + } + bm := roaring.New() + bm.FromDense(bits[:], true) + return roaring.AddOffset64(bm, int64(s.slab)<>indexSlabShift != live.slab { + var err error + if live, err = x.rotate(live); err != nil { + return err + } + } + for _, k := range keys { + live.add(k, id) + } + } + return nil +} + +// rotate waits for the last seal, then starts the next slab and seals the +// full one in the background. +func (x *hotIndex) rotate(full *hotSlab) (*hotSlab, error) { + if err := <-x.sealed; err != nil { + x.sealed <- err // final: every later rotate fails the same way + return nil, err + } + next := newHotSlab(full.slab + 1) + x.view.Store(&hotIndexView{live: next, prev: full}) + go func() { + err := x.seal(full) + if err == nil { + // The writer rotates again only after it takes this outcome. + x.view.Store(&hotIndexView{live: next}) + } + x.sealed <- err + }() + return next, nil +} + +// settle waits for the seal in flight, if any, and returns its outcome. +func (x *hotIndex) settle() error { + err := <-x.sealed + x.sealed <- err + return err +} + +// lookup returns each key's event ids in the slabs the window touches, and +// the id range those slabs span. +func (x *hotIndex) lookup(ctx context.Context, keys []TermKey, window IDRange) ([]*roaring.Bitmap, IDRange, error) { + out := make([]*roaring.Bitmap, len(keys)) + for i := range out { + out[i] = roaring.New() + } + if window.isEmpty() { + return out, window, nil + } + first, last := window.Start>>indexSlabShift, (window.End-1)>>indexSlabShift + covered := IDRange{Start: first << indexSlabShift, End: math.MaxUint32} + if last+1 < 1<<(32-indexSlabShift) { + covered.End = (last + 1) << indexSlabShift + } + + v := x.view.Load() + for slab := first; slab <= min(last, v.live.slab); slab++ { + if err := ctx.Err(); err != nil { + return nil, IDRange{}, err + } + var inMemory *hotSlab + switch { + case slab == v.live.slab: + inMemory = v.live + case v.prev != nil && slab == v.prev.slab: + inMemory = v.prev + } + for i, k := range keys { + var bm *roaring.Bitmap + if inMemory != nil { + bm = inMemory.bitmap(k) + } else { + var err error + if bm, err = x.sealedBitmap(slab, k); err != nil { + return nil, IDRange{}, err + } + } + if bm != nil { + out[i].Or(bm) + } + } + } + return out, covered, nil +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index_test.go new file mode 100644 index 000000000..b36615409 --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_index_test.go @@ -0,0 +1,339 @@ +package event + +import ( + "bytes" + "context" + "encoding/binary" + "math" + "math/rand" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + + "github.com/RoaringBitmap/roaring/v2" + "github.com/stretchr/testify/require" + + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/chunk" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/rocksdb" +) + +func randomTermKey(rng *rand.Rand) TermKey { + var k TermKey + rng.Read(k[:]) + return k +} + +// A slab filled to capacity: every event carries maxTermsPerEvent terms. +func TestHotSlab_BitmapAndTerms(t *testing.T) { + rng := rand.New(rand.NewSource(1)) + const slab = 3 + s := newHotSlab(slab) + + everyEvent := randomTermKey(rng) + // Two terms in one bucket: the same leading bytes, a different tail. + evens := randomTermKey(rng) + odds := evens + odds[len(odds)-1] ^= 0xff + + want := NewBitmaps() + sampled := map[TermKey]uint32{} + for i := range uint32(hotSlabEvents) { + id := slab< 0 { + require.Negative(t, bytes.Compare(prev[:], k[:]), "terms must ascend") + } + got := roaring.New() + require.NoError(t, got.UnmarshalBinary(serialized)) + switch { + case want[k] != nil: + require.True(t, want[k].Equals(got)) + default: + require.Equal(t, uint64(1), got.GetCardinality()) + if id, ok := sampled[k]; ok { + require.True(t, got.Contains(id)) + } + } + prev = k + terms++ + } + require.Equal(t, len(want)+hotSlabEvents*(maxTermsPerEvent-2), terms) +} + +// indexFeed feeds synthetic postings into an index and a reference: one +// term on every event, a few shared ones, and one unique to each event. +type indexFeed struct { + rng *rand.Rand + every TermKey + shared []TermKey + want Bitmaps + committed uint32 +} + +func newIndexFeed() *indexFeed { + f := &indexFeed{rng: rand.New(rand.NewSource(7)), want: NewBitmaps()} + f.every = randomTermKey(f.rng) + for range 16 { + f.shared = append(f.shared, randomTermKey(f.rng)) + } + return f +} + +func (f *indexFeed) add(t *testing.T, x *hotIndex, events int) { + t.Helper() + termKeys := make([][]TermKey, events) + for i := range termKeys { + termKeys[i] = []TermKey{f.every, f.shared[f.rng.Intn(len(f.shared))], randomTermKey(f.rng)} + for _, k := range termKeys[i] { + f.want.AddTo(k, f.committed+uint32(i)) + } + } + require.NoError(t, x.add(f.committed, termKeys)) + f.committed += uint32(events) +} + +func openIndexForTest(t *testing.T) (*hotIndex, *rocksdb.Store) { + t.Helper() + store := openRawHotChunkForTest(t, t.TempDir(), chunk.ID(0)) + t.Cleanup(func() { _ = store.Close() }) + return newHotIndex(store, 0), store +} + +func requireSealedSlabs(t *testing.T, store *rocksdb.Store, want uint32) { + t.Helper() + sealed, err := sealedHotSlabs(store) + require.NoError(t, err) + require.Equal(t, want, sealed) +} + +func TestHotIndex_AnswersAcrossSealedAndLiveSlabs(t *testing.T) { + x, store := openIndexForTest(t) + f := newIndexFeed() + // Ledgers that do not divide a slab, so slabs end mid-ledger. + for f.committed < 3*hotSlabEvents+hotSlabEvents/3 { + f.add(t, x, 7001) + } + require.NoError(t, x.settle()) + requireSealedSlabs(t, store, 3) + require.Nil(t, x.view.Load().prev, "a sealed slab leaves memory") + + keys := append([]TermKey{f.every, randomTermKey(f.rng)}, f.shared...) + for k := range f.want { + if len(keys) == 64 { + break + } + keys = append(keys, k) + } + + const slab = hotSlabEvents + const lastSlab = math.MaxUint32 / slab + windows := []struct { + name string + window IDRange + covered IDRange + }{ + {"whole chunk", IDRange{0, f.committed}, IDRange{0, 4 * slab}}, + {"first two slabs", IDRange{0, 2 * slab}, IDRange{0, 2 * slab}}, + {"inside slabs 1 and 2", IDRange{slab + 5, 3*slab - 5}, IDRange{slab, 3 * slab}}, + {"one id of slab 2", IDRange{2*slab + 9, 2*slab + 10}, IDRange{2 * slab, 3 * slab}}, + {"last sealed and live slab", IDRange{3*slab - 1, f.committed}, IDRange{2 * slab, 4 * slab}}, + {"past the live slab", IDRange{f.committed - 1, 9 * slab}, IDRange{3 * slab, 9 * slab}}, + {"last slab of the id space", IDRange{math.MaxUint32 - 5, math.MaxUint32}, IDRange{lastSlab * slab, math.MaxUint32}}, + {"empty", IDRange{40, 40}, IDRange{40, 40}}, + } + for _, w := range windows { + t.Run(w.name, func(t *testing.T) { + got, covered, err := x.lookup(context.Background(), keys, w.window) + require.NoError(t, err) + require.Equal(t, w.covered, covered) + inCovered := freshRange(uint64(covered.Start), uint64(covered.End)) + for i, k := range keys { + want := roaring.New() + if f.want[k] != nil { + want = roaring.And(f.want[k], inCovered) + } + require.True(t, want.Equals(got[i]), "term %d", i) + } + }) + } +} + +func TestHotIndex_LookupReportsUnreadableSlabs(t *testing.T) { + x, store := openIndexForTest(t) + f := newIndexFeed() + for f.committed <= hotSlabEvents { + f.add(t, x, 9000) + } + require.NoError(t, x.settle()) + requireSealedSlabs(t, store, 1) + window := IDRange{0, f.committed} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, _, err := x.lookup(ctx, []TermKey{f.every}, window) + require.ErrorIs(t, err, context.Canceled) + + garbled := randomTermKey(f.rng) + require.NoError(t, store.Put(IndexCF, hotIndexKey(0, garbled), []byte("not a bitmap"))) + _, _, err = x.lookup(context.Background(), []TermKey{garbled}, window) + require.ErrorContains(t, err, "decode events_index slab 0") + + require.NoError(t, store.Close()) + _, _, err = x.lookup(context.Background(), []TermKey{f.every}, window) + require.ErrorIs(t, err, rocksdb.ErrStoreClosed) +} + +// Lookups run while slabs fill, rotate and seal. Every committed id of the +// term on every event must be there, whichever of memory or IndexCF holds +// its slab at that instant. +func TestHotIndex_ConcurrentLookupsDuringSealing(t *testing.T) { + x, _ := openIndexForTest(t) + f := newIndexFeed() + var committed atomic.Uint32 + stop := make(chan struct{}) + var readers sync.WaitGroup + for range 3 { + readers.Go(func() { + for { + select { + case <-stop: + return + default: + } + end := committed.Load() + if end == 0 { + continue + } + got, _, err := x.lookup(context.Background(), []TermKey{f.every}, IDRange{0, end}) + if err != nil { + t.Errorf("lookup: %v", err) + return + } + if n := got[0].Rank(end - 1); n != uint64(end) { + t.Errorf("term on every event: %d of the first %d ids", n, end) + return + } + } + }) + } + for f.committed < 3*hotSlabEvents+100 { + f.add(t, x, 3001) + committed.Store(f.committed) + } + require.NoError(t, x.settle()) + close(stop) + readers.Wait() +} + +// ingestCorpus commits the corpus's events in ledgers of 9,000 until the +// chunk holds upTo of them. +func ingestCorpus(t *testing.T, h *HotStore, c *diffCorpus, upTo uint32) { + t.Helper() + for from := mustEventCount(t, h); from < upTo; from = mustEventCount(t, h) { + var payloads []Payload + for _, raw := range c.raw[from:min(from+9000, upTo)] { + payloads = append(payloads, Payload{ContractEventBytes: raw}) + } + require.NoError(t, ingestLedgerEvents(h, mustOffsets(t, h).EndLedger(), payloads)) + } +} + +// requireIndexed checks every term of the corpus over the whole chunk. +func requireIndexed(t *testing.T, h *HotStore, c *diffCorpus) { + t.Helper() + committed := freshRange(0, uint64(mustEventCount(t, h))) + for k, want := range c.index { + require.True(t, roaring.And(want, committed).Equals(lookupOne(t, h, k))) + } +} + +func TestHotStore_ReopenKeepsSealedSlabsAndIndexesTheRestAgain(t *testing.T) { + const chunkID = chunk.ID(0) + dir := t.TempDir() + corpus := newDiffCorpus(t, rand.New(rand.NewSource(3)), newDiffVocab(t), 3*hotSlabEvents+100) + + hot, raw := openHotStoreForTestAt(t, dir, chunkID) + ingestCorpus(t, hot, corpus, 2*hotSlabEvents+100) + require.NoError(t, hot.index.settle()) + requireSealedSlabs(t, raw, 2) + requireIndexed(t, hot, corpus) + require.NoError(t, raw.Close()) + + hot, raw = openHotStoreForTestAt(t, dir, chunkID) + requireSealedSlabs(t, raw, 2) + requireIndexed(t, hot, corpus) + // Ingestion carries on into the next slab. + ingestCorpus(t, hot, corpus, 3*hotSlabEvents+100) + require.NoError(t, hot.index.settle()) + requireSealedSlabs(t, raw, 3) + requireIndexed(t, hot, corpus) +} + +// A seal that fails keeps its slab answerable from memory, fails the next +// ledger's apply, and is made up for on the next open. +func TestHotStore_FailedSealIsRedoneOnReopen(t *testing.T) { + const chunkID = chunk.ID(0) + dir := t.TempDir() + corpus := newDiffCorpus(t, rand.New(rand.NewSource(4)), newDiffVocab(t), hotSlabEvents+2) + + hot, raw := openHotStoreForTestAt(t, dir, chunkID) + // A file where LoadSorted wants its directory makes every seal fail. + blocker := filepath.Join(dir, chunkID.String(), "loading") + require.NoError(t, os.WriteFile(blocker, nil, 0o600)) + ingestCorpus(t, hot, corpus, hotSlabEvents+1) + require.ErrorContains(t, hot.index.settle(), "seal index slab 0") + requireSealedSlabs(t, raw, 0) + require.Equal(t, uint32(0), hot.index.view.Load().prev.slab, "the slab stays in memory") + requireIndexed(t, hot, corpus) + + // The next ledger commits, but its apply fails. + payloads := []Payload{{ContractEventBytes: corpus.raw[hotSlabEvents+1]}} + err := ingestLedgerEvents(hot, mustOffsets(t, hot).EndLedger(), payloads) + require.ErrorContains(t, err, "seal index slab 0") + require.Equal(t, uint32(hotSlabEvents+2), countDataRows(t, raw)) + require.Equal(t, uint32(hotSlabEvents+1), mustEventCount(t, hot), "the count never runs ahead of the index") + require.NoError(t, raw.Close()) + + require.NoError(t, os.Remove(blocker)) + hot, raw = openHotStoreForTestAt(t, dir, chunkID) + requireSealedSlabs(t, raw, 1) + requireIndexed(t, hot, corpus) +} + +func countDataRows(t *testing.T, store *rocksdb.Store) uint32 { + t.Helper() + last, found, err := store.LastKey(DataCF) + require.NoError(t, err) + require.True(t, found) + return binary.BigEndian.Uint32(last) + 1 +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store.go index 266692a70..ec803094f 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store.go @@ -31,12 +31,16 @@ const ( // typically 2-3× on XDR) and read in batches via // BatchedMultiGetCF. Larger blocks give zstd more context per // compression unit and align with batch-fetch shapes. -// - IndexCF stores 20-byte (term_hash || event_id) keys with -// empty values — nothing in the values to compress, and small -// blocks reduce wasted I/O per random Lookup miss (each Lookup -// reads one block to find one key). +// - IndexCF stores one (slab || term_hash) -> bitmap entry per term of +// each sealed slab, loaded one sorted file per slab (see hotIndex). +// Auto compaction is off, and the files never overlap so it would have +// nothing to do; their index blocks sit in the block cache so a chunk's +// many files cost bounded memory. No bloom filter: it would be the +// largest thing in that cache, to save a few microseconds per slab on +// an absent term. // - OffsetsCF stores 8-byte (ledger_seq -> event_count) rows in -// the tens-of-thousands per chunk — same shape as IndexCF. +// the tens-of-thousands per chunk; small blocks waste less I/O per +// point read. const ( dataCFBlockSize = 32 * 1024 indexCFBlockSize = 4 * 1024 @@ -49,7 +53,11 @@ func hotStoreCFOptions() map[string]rocksdb.CFOptions { Compression: grocksdb.ZSTDCompression, BlockSize: dataCFBlockSize, }, - IndexCF: {BlockSize: indexCFBlockSize}, + IndexCF: { + BlockSize: indexCFBlockSize, + CacheIndexAndFilterBlocks: true, + DisableAutoCompactions: true, + }, OffsetsCF: {BlockSize: offsetsCFBlockSize}, } } @@ -63,10 +71,9 @@ func CFNames() []string { return []string{DataCF, IndexCF, OffsetsCF} } func CFOptions() map[string]rocksdb.CFOptions { return hotStoreCFOptions() } const ( - dataKeyLen = 4 // event_id (chunk encoded by per-Chunk DB directory) - indexKeyLen = 16 + 4 // term hash || event_id - offsetKeyLen = 4 // ledger_seq - offsetValLen = 4 // per-ledger event count (uint32 BE) + dataKeyLen = 4 // event_id (chunk encoded by per-Chunk DB directory) + offsetKeyLen = 4 // ledger_seq + offsetValLen = 4 // per-ledger event count (uint32 BE) ) // ErrLedgerOutOfRange is returned by IngestLedgerToBatch when the @@ -81,19 +88,19 @@ var ErrLedgerOutOfRange = errors.New("events: ledger outside chunk range") // offset chain if not rejected up front. var ErrLedgerOutOfOrder = errors.New("events: ledger out of order") -// HotStore wraps one chunk's hot RocksDB DB plus the in-memory term mirror and +// HotStore wraps one chunk's hot RocksDB DB plus the term index and // ledger-offset cache that feed the query path. // // Atomicity: the per-Chunk DB is the source of truth. IngestLedgerToBatch queues -// data + index + offsets into one atomic batch, then (post-commit) the apply -// hook updates the in-memory mirrors; warmup reconstructs them from the on-disk -// CFs on next startup. +// data + offsets into one atomic batch, then (post-commit) the apply hook +// updates the term index and the offset cache; warmup reconstructs both from +// the on-disk CFs on next startup. // // Concurrency: // // - Writes (IngestLedgerToBatch) are single-writer (one goroutine per chunk). // - Reads (LookupKeys, FetchEvents, All) take NO HotStore-level lock — they guard -// via chunkStore.IsClosed() and rely on the mirror's internal locks and +// via chunkStore.IsClosed() and rely on the index's lock-free reads and // RocksDB's thread-safety. // - Metadata split after the caller-owned store is closed: ChunkID is // infallible (cached, usable post-close); EventCount and @@ -101,7 +108,7 @@ var ErrLedgerOutOfOrder = errors.New("events: ledger out of order") type HotStore struct { chunkStore *rocksdb.Store chunkID chunk.ID - mirror *ConcurrentBitmaps + index *hotIndex offsets *ConcurrentLedgerOffsets } @@ -110,19 +117,21 @@ var _ Reader = (*HotStore)(nil) // NewWithStore wraps an ALREADY-OPEN rocksdb.Store as an events HotStore on the // three events CFs (CFNames()), running the mandatory warmup to rebuild the -// in-memory mirror + offsets. The store is owned by the caller — in production, +// term index + offsets. The store is owned by the caller — in production, // hotchunk.DB composes this facade over the shared per-chunk DB and closes that -// DB once. The store must have CFNames() registered + CFOptions() applied. -// A warmup failure returns the error WITHOUT closing the caller-owned store. +// DB once. The store must have CFNames() registered + CFOptions() applied, and +// be writable: warmup seals a full slab it indexes again as soon as the next +// one starts. A warmup failure returns the error WITHOUT closing the +// caller-owned store. func NewWithStore(store *rocksdb.Store, chunkID chunk.ID) (*HotStore, error) { - mirror, offsets, err := warmup(store, chunkID) + index, offsets, err := warmup(store, chunkID) if err != nil { return nil, fmt.Errorf("events: warmup chunk %s: %w", chunkID, err) } return &HotStore{ chunkStore: store, chunkID: chunkID, - mirror: mirror, + index: index, offsets: offsets, }, nil } @@ -168,26 +177,13 @@ func (h *HotStore) Offsets() (*LedgerOffsets, error) { return h.offsets.View(), nil } -// LookupKeys returns bitmaps for each key, aligned positionally with -// the input slice. result[i] is nil if keys[i] has no matching -// events. See Reader.LookupKeys for the semantics — in particular -// the borrowed-bitmap contract (callers must not mutate). -// -// Hot-side implementation is N in-memory mirror lookups — no I/O -// to batch — but exposing this method satisfies the Reader -// interface so callers can program against batched lookups -// uniformly. -// -// Each bitmap is a point-in-time image of its term: a sparse term is -// copied out of the mirror's published id list, a dense one is -// denseState.snapshot, the immutable clone shared with every other -// reader. Neither grows under its holder, so a walk never sees an id -// written after its lookup. +// LookupKeys returns each key's event ids in the slabs the window touches, +// aligned positionally with the input slice, and the id range those slabs +// span. See Reader.LookupKeys for the semantics. // -// The window is ignored — these images are whole-chunk and already in -// memory, and a whole term agrees with the index inside any window — so the -// covered range is the whole id space, and a query over the hot store runs -// in one stage rather than looking up ids it is already holding. +// A key with no events in the covered range gets an empty bitmap, never +// nil: nil would promise the chunk has none, and the lookup read only the +// window's slabs. Every bitmap is built for the caller. func (h *HotStore) LookupKeys( ctx context.Context, keys []TermKey, window IDRange, ) ([]*roaring.Bitmap, IDRange, error) { @@ -200,15 +196,14 @@ func (h *HotStore) LookupKeys( if len(keys) == 0 { return nil, window, nil } - results := make([]*roaring.Bitmap, len(keys)) - for i, key := range keys { - bm, err := h.mirror.Get(key) - if err != nil { - return nil, IDRange{}, fmt.Errorf("events: LookupKeys for chunk %s: %w", h.chunkID, err) - } - results[i] = bm // nil for misses — Get already returns nil bitmap for not-found + results, covered, err := h.index.lookup(ctx, keys, window) + if errors.Is(err, rocksdb.ErrStoreClosed) { + return nil, IDRange{}, stores.ErrStoreClosed } - return results, IDRange{End: math.MaxUint32}, nil + if err != nil { + return nil, IDRange{}, fmt.Errorf("events: LookupKeys for chunk %s: %w", h.chunkID, err) + } + return results, covered, nil } // FetchEvents decodes the events_data row for each provided eventID @@ -388,7 +383,7 @@ func (h *HotStore) All(ctx context.Context) iter.Seq2[Payload, error] { // contract (event IDs are assigned by arrival position). Terms are derived via // TermsForBytes on each payload's ContractEventBytes. // -// Sequence validation, before any Put or mirror mutation: +// Sequence validation, before any Put or index mutation: // // - ledgerSeq must lie within [chunkID.FirstLedger(), chunkID.LastLedger()] — // out-of-range returns ErrLedgerOutOfRange. @@ -397,14 +392,12 @@ func (h *HotStore) All(ctx context.Context) iter.Seq2[Payload, error] { // ledger is a mis-sequencing source (the ingestion loop's seq guard should // have caught it) — an error (ErrLedgerOutOfOrder), never silent tolerance. // -// Post-batch atomicity: once the batch commits, the apply hook's in-memory -// mirror + offsets updates are infallible by construction. Any failure there -// panics rather than returning an error, because a returned error would leave -// on-disk state ahead of in-memory state with no clean recovery short of -// close + reopen. +// The apply hook fails only when the index could not seal an earlier slab +// (see hotIndex.add). The ledger is committed either way, and the next open +// indexes it again, so the caller must not go on ingesting after a failure. func (h *HotStore) IngestLedgerToBatch( b *rocksdb.BatchWriter, ledgerSeq uint32, payloads []Payload, -) (func(), error) { +) (func() error, error) { // Validate BEFORE any Put. On error Store.Batch discards the whole WriteBatch, // so a mid-loop failure never orphans rows — no separate staging buffer needed. if ledgerSeq < h.chunkID.FirstLedger() || ledgerSeq > h.chunkID.LastLedger() { @@ -419,7 +412,7 @@ func (h *HotStore) IngestLedgerToBatch( } // Derive term keys per payload up front (a TermsForBytes error rejects the - // ledger without any Put) and retain them for the post-commit mirror update. + // ledger without any Put) and retain them for the post-commit index update. termKeys := make([][]TermKey, len(payloads)) for i := range payloads { keys, err := TermsForBytes(payloads[i].ContractEventBytes) @@ -445,110 +438,131 @@ func (h *HotStore) IngestLedgerToBatch( return nil, fmt.Errorf("marshal payload %d for ledger %d: %w", i, ledgerSeq, err) } scratch = blob - eventID := startID + uint32(i) - b.Put(DataCF, encodeDataKey(eventID), blob) - for _, key := range termKeys[i] { - b.Put(IndexCF, encodeIndexKey(key, eventID), nil) - } + b.Put(DataCF, encodeDataKey(startID+uint32(i)), blob) } //nolint:gosec // len bounded by the overflow guard above b.Put(OffsetsCF, encodeOffsetKey(ledgerSeq), encodeLedgerEventCount(uint32(len(payloads)))) - return func() { h.applyLedger(startID, termKeys) }, nil + return func() error { return h.applyLedger(startID, termKeys) }, nil } -// index returns the in-memory term mirror. Test-only write hook: no production -// path reads it. Kept unexported until #772 decides whether the v2 read path -// hooks into it. -func (h *HotStore) index() *ConcurrentBitmaps { return h.mirror } - -// applyLedger updates the mirror + offsets for a ledger whose rows are durable. -// Infallible by construction (IngestLedgerToBatch validated seq under the -// single-writer contract); the only non-completion is a crash, after which warmup -// rebuilds. +// applyLedger updates the index + offsets for a ledger whose rows are durable. +// It appends in memory; at a slab boundary it first waits for the previous +// slab's seal. // -// Ordering invariant: mirror BEFORE offsets. A concurrent Matches call that snapshots -// offsets then reads the mirror must see either the prior state or a consistent +// Ordering invariant: index BEFORE offsets. A concurrent Matches call that snapshots +// offsets then reads the index must see either the prior state or a consistent // later one. Reversing it would let a reader see an offsets count including IDs -// the mirror hasn't published — FetchEvents would then miss them, silently. -func (h *HotStore) applyLedger(startID uint32, termKeys [][]TermKey) { - // Batch by key so each key takes one AddTo per ledger instead of one per - // event, saving per-term lock round-trips. Cap 64 ≈ a few × unique-terms - // per ledger; the map grows past that. - perKeyIDs := make(map[TermKey][]uint32, 64) - for i, keys := range termKeys { - eventID := startID + uint32(i) - for _, key := range keys { - perKeyIDs[key] = append(perKeyIDs[key], eventID) - } - } - for key, ids := range perKeyIDs { - h.mirror.AddTo(key, ids...) +// the index hasn't published — FetchEvents would then miss them, silently. +func (h *HotStore) applyLedger(startID uint32, termKeys [][]TermKey) error { + if err := h.index.add(startID, termKeys); err != nil { + return err } //nolint:gosec // len bounded by IngestLedgerToBatch's overflow guard h.offsets.Append(uint32(len(termKeys))) + return nil } // ────────────────────────────────────────────────────────────────── -// Warmup — reconstructs the in-memory mirror + offsets from the -// per-Chunk DB's on-disk CFs. Called by NewWithStore. +// Warmup — reconstructs the term index + offsets from the per-Chunk +// DB's on-disk CFs. Called by NewWithStore. // ────────────────────────────────────────────────────────────────── -// warmup rebuilds the in-memory mirrors for chunkID by prefix-scanning -// the chunk's two on-disk caches once each: +// warmup rebuilds the in-memory state for chunkID: // -// - events_index → *ConcurrentBitmaps — every -// (TermKey, eventID) row replayed into a fresh in-memory -// bitmap mirror. // - events_offsets → *ConcurrentLedgerOffsets — every // (ledger_seq, per_ledger_count) row replayed into a fresh // offset cache. +// - events_index + events_data → *hotIndex — the sealed slabs stay +// in events_index; the events after them are indexed again from +// their events_data rows, and a full slab a later event follows is +// sealed. // // chunkID seeds ConcurrentLedgerOffsets.StartLedger for empty // chunks; on-disk rows carry the full ledger sequence themselves. -// Both mirrors are empty for fresh chunks. func warmup( chunkStore *rocksdb.Store, chunkID chunk.ID, -) (*ConcurrentBitmaps, *ConcurrentLedgerOffsets, error) { - mirror, indexUpperBound, err := warmupIndex(chunkStore) +) (*hotIndex, *ConcurrentLedgerOffsets, error) { + offsets, err := warmupOffsets(chunkStore, chunkID) if err != nil { return nil, nil, err } - offsets, err := warmupOffsets(chunkStore, chunkID) + sealed, err := sealedHotSlabs(chunkStore) if err != nil { return nil, nil, err } - if err := verifyChunkConsistency(chunkStore, offsets.TotalEvents(), indexUpperBound); err != nil { + if err := verifyChunkConsistency(chunkStore, offsets.TotalEvents(), sealed); err != nil { return nil, nil, err } - return mirror, offsets, nil + index := newHotIndex(chunkStore, sealed) + if err := replayUnsealed(chunkStore, index, sealed< 0) // and no data row at any id >= total. Together those pin the max data // id to exactly total-1 — one Get plus one bounded seek. // -// Not detected here: interior data holes (a missing id within 0..total-2, -// masked by a higher present id), under-indexed terms, and wrong +// Not detected here: interior data holes below the sealed slabs (a missing +// id masked by a higher present id), under-indexed terms, and wrong // per-ledger boundaries — each would need a full scan. The atomic batch // makes all of them impossible for the writer; an interior hole that did // appear (corruption/tamper) is caught lazily by FetchRange's short-scan -// check on first read. This is a cheap open-time tripwire on denormalized +// check on first read, or by warmup when it is among the events indexed +// again. This is a cheap open-time tripwire on denormalized // state, not load-bearing correctness. -func verifyChunkConsistency(chunkStore *rocksdb.Store, total uint32, indexUpperBound uint64) error { - // indexUpperBound is uint64 so a hostile row at eventID == - // MaxUint32 can't wrap max+1 to 0 and slip past this check. - if indexUpperBound > uint64(total) { - return fmt.Errorf("events: corrupt chunk: index references event %d but only %d committed", - indexUpperBound-1, total) +func verifyChunkConsistency(chunkStore *rocksdb.Store, total, sealedSlabs uint32) error { + if sealedSlabs > total>>indexSlabShift { + return fmt.Errorf("events: corrupt chunk: %d index slabs sealed but only %d events committed", + sealedSlabs, total) } if total > 0 { _, ok, err := chunkStore.Get(DataCF, encodeDataKey(total-1)) @@ -572,64 +586,6 @@ func verifyChunkConsistency(chunkStore *rocksdb.Store, total uint32, indexUpperB return nil } -// warmupIndex scans the events_index CF and replays every -// (TermKey, eventID) row into a fresh ConcurrentBitmaps. -// Design doc §12 step 3. -// -// Implementation: build into a single-threaded Bitmaps via -// per-term batching (rocksdb's byte-sorted iteration delivers all -// rows for term K consecutively, so a small buffer flushes when the -// term changes), then convert to ConcurrentBitmaps at the end. This -// keeps warmup off the concurrent index's per-term locking and -// promotion path — one AddTo per term instead of one per index row, -// ~50M for a 10M-event chunk. -// -// Also returns the exclusive upper bound of indexed event IDs (max + 1, -// or 0 if the index is empty; uint64 so max+1 can't wrap — see -// verifyChunkConsistency) for warmup's cross-check against the -// committed event count. -func warmupIndex(chunkStore *rocksdb.Store) (*ConcurrentBitmaps, uint64, error) { - builder := NewBitmaps() - var ( - hasPrev bool - prevTerm TermKey - buf []uint32 - indexUpperBound uint64 // max indexed event ID + 1; 0 if no rows - ) - flush := func() { - if !hasPrev || len(buf) == 0 { - return - } - builder.AddTo(prevTerm, buf...) - buf = buf[:0] - } - - for entry, err := range chunkStore.Iterate(IndexCF, nil) { - if err != nil { - return nil, 0, fmt.Errorf("events: warmup scan %s: %w", IndexCF, err) - } - if len(entry.Key) != indexKeyLen { - return nil, 0, fmt.Errorf("events: warmup unexpected %s key length %d (want %d)", - IndexCF, len(entry.Key), indexKeyLen) - } - var term TermKey - copy(term[:], entry.Key[0:16]) - eventID := binary.BigEndian.Uint32(entry.Key[16:20]) - if uint64(eventID)+1 > indexUpperBound { - indexUpperBound = uint64(eventID) + 1 - } - if hasPrev && term != prevTerm { - flush() - } - prevTerm = term - hasPrev = true - buf = append(buf, eventID) - } - flush() - - return NewConcurrentBitmapsFromBitmaps(builder), indexUpperBound, nil -} - // warmupOffsets scans events_offsets and replays every (ledger_seq, // event_count) row into a fresh *ConcurrentLedgerOffsets. The // on-disk shape matches the in-memory Append input directly @@ -688,13 +644,6 @@ func encodeDataKey(eventID uint32) []byte { return key[:] } -func encodeIndexKey(term TermKey, eventID uint32) []byte { - var key [indexKeyLen]byte - copy(key[:16], term[:]) - binary.BigEndian.PutUint32(key[16:], eventID) - return key[:] -} - func encodeOffsetKey(ledgerSeq uint32) []byte { var key [offsetKeyLen]byte binary.BigEndian.PutUint32(key[:], ledgerSeq) diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store_test.go index 9fd142f5f..0819841ba 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store_test.go @@ -7,7 +7,6 @@ import ( "fmt" "iter" "path/filepath" - "sync" "testing" "github.com/RoaringBitmap/roaring/v2" @@ -145,7 +144,7 @@ func TestHotStore_FreshChunkHasEmptyState(t *testing.T) { assert.Equal(t, chunkID.FirstLedger(), mustOffsets(t, h.store).StartLedger()) } -func TestHotStore_IngestLedgerWritesAllCFs(t *testing.T) { +func TestHotStore_IngestLedgerWritesDataAndOffsets(t *testing.T) { const chunkID = chunk.ID(0) h := openHotStoreForTest(t, chunkID) @@ -160,12 +159,10 @@ func TestHotStore_IngestLedgerWritesAllCFs(t *testing.T) { require.NoError(t, decoded.Unmarshal(got)) assert.Equal(t, p.TxHash, decoded.TxHash) - // events_index row per term. - for _, key := range keys { - _, found, err := h.store.chunkStore.Get(IndexCF, encodeIndexKey(key, 0)) - require.NoError(t, err) - assert.True(t, found, "missing index row for term %x", key) - } + // events_index holds sealed slabs only, and the first slab is not full. + _, found, err = h.store.chunkStore.LastKey(IndexCF) + require.NoError(t, err) + assert.False(t, found) // events_offsets: cumulative = 1. offVal, found, err := h.store.chunkStore.Get(OffsetsCF, encodeOffsetKey(2)) @@ -173,14 +170,14 @@ func TestHotStore_IngestLedgerWritesAllCFs(t *testing.T) { require.True(t, found) assert.Equal(t, uint32(1), binary.BigEndian.Uint32(offVal)) - // In-memory mirror sees the term. + // The index sees the term. bm := lookupOne(t, h.store, keys[0]) require.NotNil(t, bm) assert.True(t, bm.Contains(0)) - // The window is ignored: a whole-chunk image covers the whole id space. + // A lookup covers the slabs its window touches. _, covered, err := h.store.LookupKeys(context.Background(), keys[:1], IDRange{Start: 0, End: 1}) require.NoError(t, err) - assert.Equal(t, everyID, covered) // IDRange{End: math.MaxUint32} + assert.Equal(t, IDRange{Start: 0, End: 1 << indexSlabShift}, covered) assert.Equal(t, uint32(1), mustEventCount(t, h.store)) } @@ -217,37 +214,27 @@ func TestHotStore_EmptyLedgerStillWritesOffsetsAndState(t *testing.T) { assert.Equal(t, uint32(0), binary.BigEndian.Uint32(val)) } -func TestHotStore_LookupReturnsImmutableSnapshot(t *testing.T) { - // Pins the dense-mode contract: HotStore.LookupKeys returns - // immutable snapshots of the live mirror. Writers (IngestLedgerEvents) - // publish new snapshots via atomic.Pointer COW; the pointer - // previously returned by LookupKeys is never mutated. A subsequent - // IngestLedgerEvents must NOT affect a previously-returned - // bitmap pointer — callers can safely retain the pointer - // across writes. +func TestHotStore_LookupResultOutlivesLaterIngest(t *testing.T) { + // HotStore.LookupKeys builds each bitmap for its caller. A later + // IngestLedgerEvents must NOT affect a previously-returned bitmap — + // callers can safely retain it across writes. const chunkID = chunk.ID(0) h := openHotStoreForTest(t, chunkID) p, keys := makePayload("snapshot") - // Promote to dense mode so we exercise the bm.Load path (sparse - // mode allocates a fresh bitmap per Get). - for i := range uint32(70) { - require.NoError(t, ingestLedgerEvents(h.store, 2+i, []Payload{p})) - } + require.NoError(t, ingestLedgerEvents(h.store, 2, []Payload{p})) first := lookupOne(t, h.store, keys[0]) cardBefore := first.GetCardinality() - // New ingest publishes a new snapshot. The old pointer must - // remain unchanged (it's the previous snapshot). - require.NoError(t, ingestLedgerEvents(h.store, 72, []Payload{p})) + require.NoError(t, ingestLedgerEvents(h.store, 3, []Payload{p})) assert.Equal(t, cardBefore, first.GetCardinality(), - "prior LookupKeys result must be an immutable snapshot — later IngestLedgerEvents must not mutate it") + "a later IngestLedgerEvents must not change a prior LookupKeys result") second := lookupOne(t, h.store, keys[0]) assert.Equal(t, cardBefore+1, second.GetCardinality(), - "subsequent LookupKeys must observe the new snapshot") + "a later LookupKeys must observe the new event") } func TestHotStore_FetchEventsRoundTrip(t *testing.T) { @@ -390,11 +377,7 @@ func TestHotStore_CloseRejectsWrites(t *testing.T) { assert.ErrorIs(t, err, stores.ErrStoreClosed) } -// TestHotStore_PostCloseReadsError pins the contract that read methods -// fail loudly after Close. Pre-fix: LookupKeys only touched the in-memory -// mirror and returned the cached bitmaps silently, even after Close had -// released chunkStore — the only "this store is gone" signal callers -// got was when they tried to FetchEvents and hit a closed RocksDB. +// Every read method fails loudly after Close. func TestHotStore_PostCloseReadsError(t *testing.T) { const chunkID = chunk.ID(0) h := openHotStoreForTest(t, chunkID) @@ -455,11 +438,11 @@ func TestHotStore_IngestLedgerEvents_DuplicateLedgerErrors(t *testing.T) { require.Len(t, got, 1) assert.Equal(t, "a", dataSym(t, got[0]), "original event must survive the rejected re-ingest") - // The rejected payload must not reach the mirror. makePayload shares + // The rejected payload must not reach the index. makePayload shares // its contract ID, type and topic count across symbols, so topic0 is // the term that tells the two payloads apart. - assert.Nil(t, lookupOne(t, h.store, topic0TermKey(t, p2)), - "the rejected payload's topic0 term must not appear in the mirror") + assert.True(t, lookupOne(t, h.store, topic0TermKey(t, p2)).IsEmpty(), + "the rejected payload's topic0 term must not appear in the index") } // TestHotStore_IngestLedgerEvents_RejectsLedgerGap pins the contract @@ -508,27 +491,6 @@ func TestHotStore_CloseIsIdempotent(t *testing.T) { assert.NoError(t, h.raw.Close()) } -func TestHotStore_ReopenRecoversState(t *testing.T) { - // Open + ingest + close + reopen + ingest. Recovery should - // reconstruct the mirror + offsets so the second writer assigns - // event IDs continuing from where the first left off. - const chunkID = chunk.ID(0) - dir := t.TempDir() - - hot1, raw1 := openHotStoreForTestAt(t, dir, chunkID) - p1, _ := makePayload("before") - require.NoError(t, ingestLedgerEvents(hot1, 2, []Payload{p1})) - require.NoError(t, raw1.Close()) - - hot2, _ := openHotStoreForTestAt(t, dir, chunkID) - - assert.Equal(t, uint32(1), mustEventCount(t, hot2), "warmup recovered offsets") - - p2, _ := makePayload("after") - require.NoError(t, ingestLedgerEvents(hot2, 3, []Payload{p2})) - assert.Equal(t, uint32(2), mustEventCount(t, hot2)) -} - func TestHotStore_SatisfiesReader(t *testing.T) { // Compile-time guard already enforces this via the package-level // var declaration; this test demonstrates callers can hold a @@ -538,42 +500,6 @@ func TestHotStore_SatisfiesReader(t *testing.T) { assert.Equal(t, chunk.ID(0), r.ChunkID()) } -func TestHotStore_ConcurrentIngestAndLookup(t *testing.T) { - // Smoke test under -race: drive ingest on one goroutine and - // Lookup on another for a few hundred iterations. Catches - // missing locks at the IngestLedgerEvents / Lookup boundary. - const chunkID = chunk.ID(0) - h := openHotStoreForTest(t, chunkID) - - p, keys := makePayload("concurrent") - const N = 200 - - var wg sync.WaitGroup - wg.Add(2) - go func() { - defer wg.Done() - for i := range uint32(N) { - if err := ingestLedgerEvents(h.store, 2+i, []Payload{p}); err != nil { - t.Errorf("ingest %d: %v", i, err) - return - } - } - }() - go func() { - defer wg.Done() - for range N { - // A miss during the race window (writer hasn't ingested - // yet) is a nil bitmap, not an error — any error is a bug. - if _, _, err := h.store.LookupKeys(context.Background(), keys[:1], everyID); err != nil { - t.Errorf("lookup: %v", err) - return - } - } - }() - wg.Wait() - assert.Equal(t, uint32(N), mustEventCount(t, h.store)) -} - // fetchRangePayloads fully drains FetchRange into a slice for tests // that want to compare against a known sequence. Shared with // cold_reader_test.go via the package. @@ -609,11 +535,11 @@ func firstIterError(seq iter.Seq2[Payload, error]) error { return nil } -// lookupOne resolves a single term through the batched LookupKeys API -// and returns its bitmap (nil on a clean miss). Shared by the -// hot/cold/query tests that assert on one term at a time. It requires -// LookupKeys to succeed, so closed/corrupt-path tests must call -// LookupKeys directly and assert on the error. +// lookupOne resolves a single term over the whole chunk through the batched +// LookupKeys API and returns its bitmap. Shared by the hot/cold/query tests +// that assert on one term at a time. It requires LookupKeys to succeed, so +// closed/corrupt-path tests must call LookupKeys directly and assert on the +// error. func lookupOne(t *testing.T, r Reader, key TermKey) *roaring.Bitmap { t.Helper() bms, _, err := r.LookupKeys(context.Background(), []TermKey{key}, everyID) @@ -714,6 +640,15 @@ func mustOffsets(t *testing.T, r Reader) *LedgerOffsets { return o } +// addPostings injects postings into the slab being filled, the way a term +// hash collision would. No ingest may run beside it. +func addPostings(h *HotStore, key TermKey, ids ...uint32) { + live := h.index.view.Load().live + for _, id := range ids { + live.add(key, id) + } +} + // ingestLedgerEvents commits one ledger's events through IngestLedgerToBatch in // a test-owned batch and runs the post-commit apply hook — the production // write shape, reduced to a test seeding call. @@ -721,7 +656,7 @@ func ingestLedgerEvents(h *HotStore, ledgerSeq uint32, payloads []Payload) error if h.chunkStore.IsClosed() { return stores.ErrStoreClosed } - var apply func() + var apply func() error if err := h.chunkStore.Batch(func(b *rocksdb.BatchWriter) error { a, aerr := h.IngestLedgerToBatch(b, ledgerSeq, payloads) apply = a @@ -729,8 +664,5 @@ func ingestLedgerEvents(h *HotStore, ledgerSeq uint32, payloads []Payload) error }); err != nil { return err } - if apply != nil { - apply() - } - return nil + return apply() } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_warmup_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_warmup_test.go index 64e58f836..5f84d0546 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_warmup_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_warmup_test.go @@ -1,8 +1,6 @@ package event import ( - "fmt" - "math" "testing" "github.com/stretchr/testify/assert" @@ -17,89 +15,20 @@ import ( // document the "fresh chunk → empty caches", "ingested chunk → // reconstructed caches" contract. -func TestWarmup_FreshChunkProducesEmptyMirrorsViaNewWithStore(t *testing.T) { +func TestWarmup_FreshChunkProducesEmptyStateViaNewWithStore(t *testing.T) { const chunkID = chunk.ID(0) h := openHotStoreForTest(t, chunkID) - // A fresh mirror is empty: probing any term is a clean miss - // (nil bitmap, no error). - bm, err := h.store.mirror.Get(ComputeTermKey([]byte("any"), FieldContractID)) - require.NoError(t, err) - assert.Nil(t, bm) + // A fresh index is empty: probing any term is a clean miss, an empty + // bitmap rather than nil. + bm := lookupOne(t, h.store, ComputeTermKey([]byte("any"), FieldContractID)) + require.NotNil(t, bm) + assert.True(t, bm.IsEmpty()) assert.Zero(t, h.store.offsets.LedgerCount()) assert.Equal(t, uint32(0), h.store.offsets.TotalEvents()) assert.Equal(t, chunkID.FirstLedger(), h.store.offsets.StartLedger()) } -func TestWarmup_RebuildsMirrorFromIngestedRows(t *testing.T) { - // Ingest into one HotStore, close, reopen; the reopened store's - // mirror must hold the same terms (warmup replayed them from - // events_index). One-directional by design: every seeded term - // must survive with the right cardinality, but a phantom extra - // term in the reopened mirror would pass unnoticed — - // ConcurrentBitmaps deliberately has no term enumeration to - // check the other direction with. - const chunkID = chunk.ID(0) - dir := t.TempDir() - - hot1, raw1 := openHotStoreForTestAt(t, dir, chunkID) - p1, _ := makePayload("alpha") - p2, _ := makePayload("beta") - require.NoError(t, ingestLedgerEvents(hot1, 2, []Payload{p1, p2})) - - // Derive the terms each seeded payload contributes (the same - // TermsForBytes the ingest path indexed by), then snapshot - // their pre-close cardinalities straight from the live mirror. - var seededKeys []TermKey - for _, p := range []Payload{p1, p2} { - keys, err := TermsForBytes(p.ContractEventBytes) - require.NoError(t, err) - seededKeys = append(seededKeys, keys...) - } - expected := make(map[TermKey]uint64) - for _, k := range seededKeys { - bm, err := hot1.mirror.Get(k) - require.NoError(t, err) - require.NotNil(t, bm, "seeded term missing from the pre-close mirror") - expected[k] = bm.GetCardinality() - } - require.NoError(t, raw1.Close()) - - // Reopen — warmup replays events_index into a fresh mirror. - hot2, _ := openHotStoreForTestAt(t, dir, chunkID) - - got := make(map[TermKey]uint64) - for k := range expected { - bm, err := hot2.mirror.Get(k) - require.NoError(t, err) - require.NotNil(t, bm, "seeded term missing from the reopened mirror") - got[k] = bm.GetCardinality() - } - assert.Equal(t, expected, got) -} - -func TestWarmup_RestoresEventIDsForRepeatedTerm(t *testing.T) { - const chunkID = chunk.ID(0) - dir := t.TempDir() - - hot1, raw1 := openHotStoreForTestAt(t, dir, chunkID) - p1, _ := makePayload("shared") - p2, _ := makePayload("shared") - p3, _ := makePayload("shared") - require.NoError(t, ingestLedgerEvents(hot1, 2, []Payload{p1, p2, p3})) - require.NoError(t, raw1.Close()) - - hot2, _ := openHotStoreForTestAt(t, dir, chunkID) - - contractTermKey := ComputeTermKey(eventOf(p1).ContractId[:], FieldContractID) - bm := lookupOne(t, hot2, contractTermKey) - require.NotNil(t, bm) - assert.Equal(t, uint64(3), bm.GetCardinality()) - for id := range uint32(3) { - assert.True(t, bm.Contains(id), "warmup missed event id %d", id) - } -} - func TestWarmup_OffsetsReconstructedAcrossLedgers(t *testing.T) { const chunkID = chunk.ID(0) dir := t.TempDir() @@ -223,46 +152,35 @@ func TestWarmup_RejectsOrphanInEmptyChunk(t *testing.T) { require.ErrorContains(t, err, "data present at id >= committed count 0") } -func TestWarmup_RejectsMissingTailDataEvent(t *testing.T) { - const chunkID = chunk.ID(0) - dir := t.TempDir() - - hot1, raw1 := openHotStoreForTestAt(t, dir, chunkID) - p1, _ := makePayload("a") - p2, _ := makePayload("b") - require.NoError(t, ingestLedgerEvents(hot1, 2, []Payload{p1, p2})) // total = 2 - require.NoError(t, raw1.Close()) - - // Drop the last data row (event id total-1 == 1) while offsets still - // count 2. - corruptHotChunk(t, dir, chunkID, func(raw *rocksdb.Store) { - require.NoError(t, raw.Delete(DataCF, encodeDataKey(1))) - }) - - _, _, err := tryOpenHotStoreForTest(t, dir, chunkID) - require.ErrorContains(t, err, "missing from data") -} - -func TestWarmup_RejectsIndexBeyondCommitted(t *testing.T) { - const chunkID = chunk.ID(0) - dir := t.TempDir() - - hot1, raw1 := openHotStoreForTestAt(t, dir, chunkID) - p1, _ := makePayload("a") - p2, _ := makePayload("b") - require.NoError(t, ingestLedgerEvents(hot1, 2, []Payload{p1, p2})) // total = 2 - require.NoError(t, raw1.Close()) - - // An index row at exactly total (id 2): the tightest "beyond - // committed" case, pinning the > (not >=) bound — valid ids are 0..1. - corruptHotChunk(t, dir, chunkID, func(raw *rocksdb.Store) { - var term TermKey - term[0] = 0x99 - require.NoError(t, raw.Put(IndexCF, encodeIndexKey(term, 2), nil)) - }) - - _, _, err := tryOpenHotStoreForTest(t, dir, chunkID) - require.ErrorContains(t, err, "index references event 2 but only 2 committed") +// A data row the offsets count must be there: the last one is checked +// directly, and the unsealed ones as warmup indexes them again. +func TestWarmup_RejectsMissingDataEvent(t *testing.T) { + for _, tc := range []struct { + name string + missing uint32 + want string + }{ + {"the last event", 1, "missing from data"}, + {"an earlier unsealed event", 0, "where event 0 belongs"}, + } { + t.Run(tc.name, func(t *testing.T) { + const chunkID = chunk.ID(0) + dir := t.TempDir() + + hot1, raw1 := openHotStoreForTestAt(t, dir, chunkID) + p1, _ := makePayload("a") + p2, _ := makePayload("b") + require.NoError(t, ingestLedgerEvents(hot1, 2, []Payload{p1, p2})) // total = 2 + require.NoError(t, raw1.Close()) + + corruptHotChunk(t, dir, chunkID, func(raw *rocksdb.Store) { + require.NoError(t, raw.Delete(DataCF, encodeDataKey(tc.missing))) + }) + + _, _, err := tryOpenHotStoreForTest(t, dir, chunkID) + require.ErrorContains(t, err, tc.want) + }) + } } func TestWarmup_OffsetsHandleEmptyTrailingLedger(t *testing.T) { @@ -286,11 +204,7 @@ func TestWarmup_OffsetsHandleEmptyTrailingLedger(t *testing.T) { assert.Equal(t, uint32(1), end, "empty ledger reports zero-width range") } -// TestWarmup_RejectsIndexRowAtMaxUint32 pins the tripwire against its -// wrap-adversarial input: an index row at eventID == MaxUint32. With a -// uint32 upper bound, max+1 wraps to 0 and slips past the > total check -// — the exact tamper the tripwire exists to catch; the bound is uint64. -func TestWarmup_RejectsIndexRowAtMaxUint32(t *testing.T) { +func TestWarmup_RejectsSealedSlabBeyondOffsets(t *testing.T) { const chunkID = chunk.ID(0) dir := t.TempDir() @@ -300,12 +214,9 @@ func TestWarmup_RejectsIndexRowAtMaxUint32(t *testing.T) { require.NoError(t, raw1.Close()) corruptHotChunk(t, dir, chunkID, func(raw *rocksdb.Store) { - var term TermKey - term[0] = 0x99 - require.NoError(t, raw.Put(IndexCF, encodeIndexKey(term, math.MaxUint32), nil)) + require.NoError(t, raw.Put(IndexCF, hotIndexKey(0, TermKey{}), nil)) }) _, _, err := tryOpenHotStoreForTest(t, dir, chunkID) - require.ErrorContains(t, err, - fmt.Sprintf("index references event %d but only 1 committed", uint32(math.MaxUint32))) + require.ErrorContains(t, err, "1 index slabs sealed but only 1 events committed") } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/index_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/index_test.go index 4a87bbde2..5e2585877 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/index_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/index_test.go @@ -250,7 +250,7 @@ func TestTermsForBytes_TopicCountClippedToMax(t *testing.T) { keys, err := TermsForBytes(marshaledEvent(t, ev)) require.NoError(t, err) - assert.Len(t, keys, 3+protocol.MaxTopicCount, + assert.Len(t, keys, maxTermsPerEvent, "type + contract-ID + topic-count terms, then MaxTopicCount topic terms (extras dropped)") } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/match_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/match_test.go index 2597fcecf..889ef6e2f 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/match_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/match_test.go @@ -302,27 +302,6 @@ func TestQuery_DuplicateTermsAcrossFiltersDedupedInLookup(t *testing.T) { assert.Equal(t, 3, cr.totalKeys, "Query must dedupe the shared topic0=alpha term") } -func TestQuery_DoesNotMutateMirrorBitmaps(t *testing.T) { - fx := newQueryFixture(t) - // Snapshot the mirror's bitmap for topic0=alpha before any query. - key := ComputeTermKey(fx.t0aRaw, FieldTopic0) - before := lookupOne(t, fx.store, key) - beforeCard := before.GetCardinality() - - // Run several queries that all touch the topic0=alpha term. - for range 3 { - _, err := Query(context.Background(), fx.store, []Filter{ - {ContractID: fx.contractA[:], Topics: [protocol.MaxTopicCount][]byte{fx.t0aRaw}}, - {ContractID: fx.contractB[:], Topics: [protocol.MaxTopicCount][]byte{fx.t0aRaw}}, - }, QueryOptions{Range: wholeChunk(t, fx.store)}) - require.NoError(t, err) - } - - after := lookupOne(t, fx.store, key) - assert.Equal(t, beforeCard, after.GetCardinality(), - "Query must not mutate the mirror's bitmaps") -} - func TestQuery_CanceledContextReturnsError(t *testing.T) { fx := newQueryFixture(t) ctx, cancel := context.WithCancel(context.Background()) @@ -574,15 +553,14 @@ func TestQuery_MixedSuccessFilterList(t *testing.T) { fx := newQueryFixture(t) // Construct a topic that no fixture event uses, so its term key - // has no entry in the index (LookupKeys returns nil for it). + // has no entry in the index. missingSym := xdr.ScSymbol("nonexistent") missingTopic := xdr.ScVal{Type: xdr.ScValTypeScvSymbol, Sym: &missingSym} missingRaw, err := missingTopic.MarshalBinary() require.NoError(t, err) - // Sanity check: this term really isn't in the index. LookupKeys - // (which Query uses) signals the miss with a nil slot. + // Sanity check: this term really isn't in the index. missingKey := ComputeTermKey(missingRaw, FieldTopic0) - require.Nil(t, lookupOne(t, fx.store, missingKey), + require.True(t, lookupOne(t, fx.store, missingKey).IsEmpty(), "fixture sanity: 'nonexistent' must not be indexed") got, err := Query(context.Background(), fx.store, []Filter{ @@ -1162,7 +1140,7 @@ func TestQuery_InvalidFilterRejected(t *testing.T) { // - descending + range + cap → the slab walk run high to low // // What we don't replay against cold: -// - The mirror-poisoning collision test (mutating an mmap'd cold +// - The index-poisoning collision test (mutating an mmap'd cold // index.pack would require writing a malformed fixture; the post- // filter itself is the same code regardless of which Reader fed it). // - Per-call options validation (negative MaxEvents, short ContractID) @@ -1593,7 +1571,7 @@ func TestMatches_DropsAreInvisible(t *testing.T) { // and every candidate after the true match drops. gammaKey := ComputeTermKey(fx.t0cRaw, FieldTopic1) for _, fp := range []uint32{2, 3, 4} { - fx.store.index().AddTo(gammaKey, fp) + addPostings(fx.store, gammaKey, fp) } filters := []Filter{{Topics: [protocol.MaxTopicCount][]byte{nil, fx.t0cRaw}}} @@ -1637,39 +1615,6 @@ func TestMatches_EmptyStreams(t *testing.T) { wholeChunk(t, fx.store), false)) } -// TestMatches_LeavesSharedSnapshotUntouched pins that a narrowing window over -// a single-term filter leaves the hot mirror's shared bitmap as it was. Only -// dense terms are shared, so the term is first promoted with injected ids -// above the query window. -func TestMatches_LeavesSharedSnapshotUntouched(t *testing.T) { - fx := newQueryFixture(t) - key := ComputeTermKey(fx.contractA[:], FieldContractID) - // Promote contract A's term (real matches: ids 0, 1, 4) to dense - // mode. The injected ids sit above the chunk's EventCount and every - // window below, so they are clipped before any fetch. - for id := uint32(100); id < 200; id++ { - fx.store.index().AddTo(key, id) - } - before := lookupOne(t, fx.store, key) - require.GreaterOrEqual(t, before.GetCardinality(), uint64(100), - "fixture sanity: the term must be dense so LookupKeys borrows") - snapshot := before.Clone() - - // A narrowing window over a single-term filter. - got := collectMatches(t, fx.store, []Filter{{ContractID: fx.contractA[:]}}, - IDRange{Start: 0, End: 2}, false) - assert.Equal(t, []uint32{0, 1}, matchOrdinals(got)) - - after := lookupOne(t, fx.store, key) - assert.True(t, snapshot.Equals(after), - "Matches must not mutate the mirror's shared term bitmap") - - // The same filter over the whole chunk still sees every id. - full := collectMatches(t, fx.store, []Filter{{ContractID: fx.contractA[:]}}, - wholeChunk(t, fx.store), false) - assert.Equal(t, []uint32{0, 1, 4}, matchOrdinals(full)) -} - // TestQuery_PostFilterRejectsTermHashCollision pins the defensive // post-filter: if a bitmap entry survives the index lookup but the // underlying event's bytes don't actually match the filter clause, @@ -1678,9 +1623,9 @@ func TestMatches_LeavesSharedSnapshotUntouched(t *testing.T) { // collision (or a corrupt index) could otherwise leak the wrong // event through Query. // -// We force the case by injecting a false-positive entry directly -// into the mirror's bitmap for the "topic1 == gamma" term, -// equivalent to what a real collision would produce. +// We force the case by injecting a false-positive posting directly +// into the hot index for the "topic1 == gamma" term, equivalent to +// what a real collision would produce. func TestQuery_PostFilterRejectsTermHashCollision(t *testing.T) { fx := newQueryFixture(t) @@ -1698,10 +1643,7 @@ func TestQuery_PostFilterRejectsTermHashCollision(t *testing.T) { require.NoError(t, err) require.Len(t, want, 1, "fixture sanity: exactly one true match before injection") - // ConcurrentBitmaps.AddTo is the writer-side API the ingest path uses - // to register (term, eventID) pairs. No concurrent ingest is running - // in this test, so the single-writer contract is satisfied. - fx.store.index().AddTo(gammaKey, 4) + addPostings(fx.store, gammaKey, 4) after := lookupOne(t, fx.store, gammaKey) require.True(t, after.Contains(4), "fixture sanity: collision id=4 is now in the bitmap") @@ -1720,10 +1662,7 @@ func TestQuery_PostFilterRejectsTermHashCollision(t *testing.T) { // match carries its own ordinal when a dropped candidate is at a // LOWER ordinal in the same batch. Survivor-index alignment (the // natural rewrite mistake, ids[len(out)] instead of ids[i]) would hand -// the dropped candidate's ordinal to the true match. This is a direct -// unit test because the scenario cannot be staged through the hot -// mirror: AddTo only accepts ascending ids, so an injected false -// positive can never precede a real match, but a genuine xxh3 +// the dropped candidate's ordinal to the true match. A genuine xxh3 // collision can sit at any ordinal, so the alignment is load-bearing. func TestPostFilter_OrdinalAlignmentWithLeadingDrop(t *testing.T) { var cid xdr.ContractId diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/matches_differential_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/matches_differential_test.go index 85463f68e..2f8d367f1 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/matches_differential_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/matches_differential_test.go @@ -1,13 +1,15 @@ package event // The in-memory chunk the match-path tests run against, served through -// LookupKeys, and the borrow-safety gate over it. +// LookupKeys, and the concurrent-ingest gate that checks a hot store +// against it. import ( "context" "errors" "iter" "math/rand" + "slices" "testing" "github.com/RoaringBitmap/roaring/v2" @@ -21,8 +23,8 @@ import ( // diffCorpus is an in-memory chunk with one distinct event per id. type diffCorpus struct { - raw [][]byte - mirror *ConcurrentBitmaps + raw [][]byte + index Bitmaps } // diffReader serves the corpus through LookupKeys, the seam Matches reads the @@ -45,11 +47,7 @@ func (r diffReader) LookupKeys( ) ([]*roaring.Bitmap, IDRange, error) { out := make([]*roaring.Bitmap, len(keys)) for i, k := range keys { - bm, err := r.c.mirror.Get(k) - if err != nil { - return nil, IDRange{}, err - } - out[i] = bm + out[i] = r.c.index[k] } return out, window, nil } @@ -119,7 +117,7 @@ func newDiffVocab(tb testing.TB) *diffVocab { func newDiffCorpus(t *testing.T, rng *rand.Rand, v *diffVocab, n int) *diffCorpus { t.Helper() - c := &diffCorpus{mirror: NewConcurrentBitmapsFromBitmaps(NewBitmaps())} + c := &diffCorpus{index: NewBitmaps()} for id := range n { var cid xdr.ContractId copy(cid[:], v.contracts[rng.Intn(len(v.contracts))]) @@ -146,7 +144,7 @@ func newDiffCorpus(t *testing.T, rng *rand.Rand, v *diffVocab, n int) *diffCorpu keys, err := TermsForBytes(raw) require.NoError(t, err) for _, k := range keys { - c.mirror.AddTo(k, uint32(id)) + c.index.AddTo(k, uint32(id)) } } return c @@ -197,81 +195,52 @@ func collectOrdinals(t *testing.T, r Reader, filters []Filter, w IDRange, desc b return out } -// The match path holds mirror snapshots across a whole walk while AddTo -// publishes new termStates on the same keys, sparse-to-dense promotion -// included. Under -race any write reaching a held snapshot fails the run; -// without it, the identity check pins that a pinned window ignores later ingest. -func TestMatches_ConcurrentIngestBorrowSafety(t *testing.T) { +// Matches runs beside ingestion that fills, rotates and seals slabs. A walk +// over the events committed when it started must find exactly the matches +// among them: an id the offsets count is already indexed, wherever its slab +// is held, and nothing ingested later leaks in. +func TestMatches_FindsEveryCommittedEventDuringIngest(t *testing.T) { rng := rand.New(rand.NewSource(20260830)) v := newDiffVocab(t) - const corpusSize = 400 - const pinned = corpusSize / 2 - - corpus := &diffCorpus{mirror: NewConcurrentBitmapsFromBitmaps(NewBitmaps())} - keysByID := make([][]TermKey, corpusSize) - for id := range corpusSize { - var cid xdr.ContractId - copy(cid[:], v.contracts[rng.Intn(len(v.contracts))]) - topics := make([]xdr.ScVal, 0, 3) - for range 1 + rng.Intn(3) { - topics = append(topics, v.topics[rng.Intn(len(v.topics))]) - } - sym := xdr.ScSymbol("data") - ev := xdr.ContractEvent{ - ContractId: &cid, - Type: v.types[rng.Intn(len(v.types))], - Body: xdr.ContractEventBody{ - V: 0, - V0: &xdr.ContractEventV0{ - Topics: topics, - Data: xdr.ScVal{Type: xdr.ScValTypeScvSymbol, Sym: &sym}, - }, - }, - } - raw, err := ev.MarshalBinary() - require.NoError(t, err) - corpus.raw = append(corpus.raw, raw) - keys, err := TermsForBytes(raw) - require.NoError(t, err) - keysByID[id] = keys - } - // Only the pinned window is indexed up front; the writer feeds the rest - // live, and most keys cross the promotion threshold mid-run. - for id := range pinned { - for _, k := range keysByID[id] { - corpus.mirror.AddTo(k, uint32(id)) - } - } - - r := diffReader{corpus} + const ( + ledgers = 16 + eventsPerLedger = 5000 // more than one slab in all + ) + corpus := newDiffCorpus(t, rng, v, ledgers*eventsPerLedger) et := xdr.ContractEventTypeContract filters := []Filter{ - {ContractID: v.contracts[0]}, - {Topics: [protocol.MaxTopicCount][]byte{0: v.topicRaw[1]}, EventType: &et}, - {TopicCount: TopicCountFilter{Count: 2}}, + {ContractID: v.contracts[0], Topics: [protocol.MaxTopicCount][]byte{0: v.topicRaw[1]}}, + {ContractID: v.contracts[1], TopicCount: TopicCountFilter{Count: 3}, EventType: &et}, } - window := IDRange{Start: 0, End: pinned} - want := collectOrdinals(t, r, filters, window, false) - require.NotEmpty(t, want, "fixture sanity: the pinned window must match something") + want := collectOrdinals(t, diffReader{corpus}, filters, IDRange{End: uint32(len(corpus.raw))}, false) + require.NotEmpty(t, want, "fixture sanity: the filters must match something") + h := openHotStoreForTest(t, chunk.ID(0)) done := make(chan struct{}) go func() { defer close(done) - for id := pinned; id < corpusSize; id++ { - for _, k := range keysByID[id] { - corpus.mirror.AddTo(k, uint32(id)) + for l := range ledgers { + payloads := make([]Payload, eventsPerLedger) + for i := range payloads { + payloads[i] = Payload{ContractEventBytes: corpus.raw[l*eventsPerLedger+i]} + } + if err := ingestLedgerEvents(h.store, chunk.ID(0).FirstLedger()+uint32(l), payloads); err != nil { + t.Errorf("ingest ledger %d: %v", l, err) + return } } }() - for { + for ingesting := true; ingesting; { select { case <-done: - require.Equal(t, want, collectOrdinals(t, r, filters, window, false), - "pinned window changed after ingest completed") - return + ingesting = false default: - require.Equal(t, want, collectOrdinals(t, r, filters, window, false), - "pinned window changed mid-ingest") } + committed := mustEventCount(t, h.store) + n, _ := slices.BinarySearch(want, committed) + got := collectOrdinals(t, h.store, filters, IDRange{End: committed}, false) + require.Equal(t, append([]uint32(nil), want[:n]...), got, "over the first %d events", committed) } + require.NoError(t, h.store.index.settle()) + requireSealedSlabs(t, h.raw, 1) } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go index 6afdf8401..158bc06d5 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go @@ -83,7 +83,7 @@ type Reader interface { // - ColdReader returns the lazily-decoded LedgerOffsets cached // on the reader; the same pointer is returned to every // caller. Both paths must treat the returned value as - // read-only — mutation would corrupt either the live mirror + // read-only — mutation would corrupt either the live offsets // (hot, indirectly via the view's backing slice) or every // other reader holding the cached pointer (cold). // @@ -92,8 +92,10 @@ type Reader interface { // LookupKeys returns bitmaps for each key, aligned positionally // with the input slice (result[i] corresponds to keys[i]). - // result[i] is nil if keys[i] has no matching events in this - // chunk — a per-key miss is not an error. + // result[i] is nil only when the reader knows keys[i] has no + // events anywhere in this chunk; otherwise it is a bitmap, empty + // included, that answers for the covered range. A per-key miss + // is not an error. // // window is the id range the caller asks to be answered for. // The second result is the range the answer actually covers: it @@ -112,21 +114,15 @@ type Reader interface { // ColdReader coalesces the underlying packfile reads into a // single ReadItems pass, fanning out across the worker count // configured via ColdReaderOptions.Concurrency. It reads a split term - // only in the slabs the window reaches, and covers them. HotStore returns - // snapshots of the live mirror shared by all readers of a term; - // a dense term written since its last lookup is cloned once, by - // the first reader to look it up, and that clone is then shared, - // window or no window. + // only in the slabs the window reaches, and covers them. HotStore reads + // the slabs the window reaches, from memory or from its index column + // family, and covers them. // - // Callers MUST treat returned bitmaps as read-only. Dense hot - // snapshots are shared with other readers; sparse hot terms - // return a fresh caller-owned bitmap per lookup; cold-path - // bitmaps are freshly unmarshaled and owned by the caller. See - // ConcurrentBitmaps.Get. + // Callers MUST treat returned bitmaps as read-only. // // ctx cancels in-flight I/O on the cold path (MPHF load, // index.pack ReadAt); hot side checks ctx as a fast guard before - // touching the in-memory mirror. + // touching the index. LookupKeys(ctx context.Context, keys []TermKey, window IDRange) ([]*roaring.Bitmap, IDRange, error) // FetchEvents decodes events for the supplied chunk-relative diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/roaring_contract_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/roaring_contract_test.go index 25d637d37..29737c4d3 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/roaring_contract_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/roaring_contract_test.go @@ -1,12 +1,11 @@ package event -// roaring_contract_test.go pins the roaring properties the event index rests -// on. ConcurrentBitmaps.Get publishes one bitmap to every concurrent reader -// and never mutates it, and the match path hands those bitmaps to FastAnd, -// NextValue and PreviousValue from many goroutines at once. That is sound -// only while FastAnd reads its arguments and returns containers that share -// no storage with them, and the searches are read-only, inclusive of the -// target, and return -1 for none. The tests are written against roaring +// roaring_contract_test.go pins the roaring properties the match path rests +// on. The bitmaps Reader.LookupKeys returns are read-only, and the walk hands +// them to FastAnd, NextValue and PreviousValue. That is sound only while +// FastAnd reads its arguments and returns containers that share no storage +// with them, and the searches are read-only, inclusive of the target, and +// return -1 for none. The tests are written against roaring // alone, so they are the first thing to run on a version bump. // // Pinned version: github.com/RoaringBitmap/roaring/v2 v2.26.0. @@ -14,18 +13,16 @@ package event import ( "math/rand" "slices" - "sync" "testing" "github.com/RoaringBitmap/roaring/v2" "github.com/stretchr/testify/require" ) -// sharedBitmaps returns bitmaps in the shape the index publishes: the -// copy-on-write Clone that denseState.snapshot hands to readers, sharing its -// containers with the writer's bitmap. The set spans all three container -// kinds and several high keys, since the roaring paths branch on both. -func sharedBitmaps(t *testing.T) []*roaring.Bitmap { +// containerKinds returns the bitmaps the tests hand to roaring as read-only +// arguments: all three container kinds over several high keys, since the +// roaring paths branch on both. +func containerKinds(t *testing.T) []*roaring.Bitmap { t.Helper() rng := rand.New(rand.NewSource(20260909)) @@ -55,15 +52,7 @@ func sharedBitmaps(t *testing.T) []*roaring.Bitmap { } } - out := make([]*roaring.Bitmap, 0, 4) - for _, wbm := range []*roaring.Bitmap{sparse, dense, runs, wide} { - wbm.SetCopyOnWrite(true) - snap := wbm.Clone() - require.True(t, snap.GetCopyOnWrite(), - "fixture: the published snapshot must carry the copy-on-write mark") - out = append(out, snap) - } - return out + return []*roaring.Bitmap{sparse, dense, runs, wide} } func bitmapBytes(t *testing.T, bm *roaring.Bitmap) []byte { @@ -86,8 +75,8 @@ func requireUnchanged(t *testing.T, bms []*roaring.Bitmap, before [][]byte, op s t.Helper() for i, bm := range bms { require.Equal(t, before[i], bitmapBytes(t, bm), - "%s mutated shared argument %d: roaring can no longer be handed "+ - "denseState.snapshot bitmaps at this version", op, i) + "%s mutated input %d: roaring can no longer be handed "+ + "read-only bitmaps at this version", op, i) } } @@ -126,8 +115,8 @@ var windowRanges = []struct { } // TestRoaringContract_AggregationDoesNotMutateInputs pins that FastAnd -// leaves shared arguments byte-identical and equals the pairwise And chain, -// for every non-empty subset of the shared bitmaps beside the window: each +// leaves its inputs byte-identical and equals the pairwise And chain, for +// every non-empty subset of the input bitmaps beside the window: each // container kind meets the window alone, in pairs and all together, which // is every kernel a plan of one to seven terms can reach. func TestRoaringContract_AggregationDoesNotMutateInputs(t *testing.T) { @@ -136,10 +125,10 @@ func TestRoaringContract_AggregationDoesNotMutateInputs(t *testing.T) { for _, w := range windowRanges { t.Run(w.name, func(t *testing.T) { t.Parallel() - shared := sharedBitmaps(t) - for pick := 1; pick < 1<