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<