diff --git a/cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go b/cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go index 1e22c786f..71477764d 100644 --- a/cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/ingest/ingest_test.go @@ -560,10 +560,10 @@ func TestEventsColdWriter_Readback(t *testing.T) { cnt, err := cr.EventCount() require.NoError(t, err) require.Equal(t, uint32(2), cnt) - bms, err := cr.LookupKeys(context.Background(), []event.TermKey{term}) + post, err := cr.LookupKeys(context.Background(), []event.TermKey{term}) require.NoError(t, err) - require.NotNil(t, bms[0]) - require.Equal(t, uint64(2), bms[0].GetCardinality()) + require.True(t, post[0].Present()) + require.Equal(t, uint64(2), post[0].Cardinality()) } // ───────────────────────── V0 (pre-Soroban) events handling ───────────────────────── @@ -618,8 +618,8 @@ func TestEventsColdWriter_V0KeepsOffsetsContiguous(t *testing.T) { // And the event is queryable by its term. bms, err := cr.LookupKeys(context.Background(), []event.TermKey{term}) require.NoError(t, err) - require.NotNil(t, bms[0]) - require.Equal(t, uint64(1), bms[0].GetCardinality()) + require.True(t, bms[0].Present()) + require.Equal(t, uint64(1), bms[0].Cardinality()) } // TestWriteColdChunk_EventlessChunk_FullyReadable drives a full cold chunk of V0 @@ -664,7 +664,7 @@ func TestWriteColdChunk_EventlessChunk_FullyReadable(t *testing.T) { anyKey := event.ComputeTermKey([]byte("any"), event.FieldContractID) bms, lerr := cr.LookupKeys(context.Background(), []event.TermKey{anyKey}) require.NoError(t, lerr) - require.Nil(t, bms[0], "a term with no matching events misses cleanly (nil bitmap)") + require.False(t, bms[0].Present(), "a term with no matching events misses cleanly") // Metrics still fired: one aggregate per-chunk, one (clean) per-ingester. require.Equal(t, 1, sink.coldChunkTotals, "ColdChunkTotal must fire for an eventless chunk") @@ -718,7 +718,7 @@ func TestColdChunk_Success(t *testing.T) { defer func() { require.NoError(t, ecr.Close()) }() bms, err := ecr.LookupKeys(context.Background(), []event.TermKey{term}) require.NoError(t, err) - require.Equal(t, uint64(2), bms[0].GetCardinality()) + require.Equal(t, uint64(2), bms[0].Cardinality()) // Txhash .bin count. binEntries := rpcv2test.ReadColdBin(t, txhashBinPath(filepath.Join(coldDir, dataTypeTxhash))) @@ -1025,9 +1025,9 @@ func TestWriteColdChunk_ByteIdentity_SharedWalk(t *testing.T) { bms, err := ecr.LookupKeys(context.Background(), terms) require.NoError(t, err) for i, k := range terms { - require.NotNil(t, bms[i], "term %d present in reference must resolve", i) - require.Equal(t, wantTermIDs[k], bms[i].ToArray(), - "cold term bitmap must carry the same event IDs as the shaping reference") + require.True(t, bms[i].Present(), "term %d present in reference must resolve", i) + require.Equal(t, wantTermIDs[k], bms[i].Bitmap().ToArray(), + "cold term postings must carry the same event IDs as the shaping reference") } } @@ -1112,8 +1112,8 @@ func TestWriteColdChunk_EventsCold_Readback(t *testing.T) { require.Equal(t, uint32(len(evSeqs)), cnt) bms, err := cr.LookupKeys(context.Background(), []event.TermKey{term}) require.NoError(t, err) - require.NotNil(t, bms[0]) - require.Equal(t, uint64(len(evSeqs)), bms[0].GetCardinality()) + require.True(t, bms[0].Present()) + require.Equal(t, uint64(len(evSeqs)), bms[0].Cardinality()) } // ───────────────────────── drain stream errors ───────────────────────── diff --git a/cmd/stellar-rpc/internal/rpcv2/query/event_page_test.go b/cmd/stellar-rpc/internal/rpcv2/query/event_page_test.go index e514eb384..9aed4d2a5 100644 --- a/cmd/stellar-rpc/internal/rpcv2/query/event_page_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/query/event_page_test.go @@ -932,10 +932,12 @@ func (f *fakeEventReader) EventCount() (uint32, error) { func (f *fakeEventReader) Offsets() (*event.LedgerOffsets, error) { return f.ofs, nil } -func (f *fakeEventReader) LookupKeys(_ context.Context, keys []event.TermKey) ([]*roaring.Bitmap, error) { - out := make([]*roaring.Bitmap, len(keys)) +func (f *fakeEventReader) LookupKeys(_ context.Context, keys []event.TermKey) ([]event.Postings, error) { + out := make([]event.Postings, len(keys)) for i, k := range keys { - out[i] = f.bitmaps[k] + // An absent key maps to a nil bitmap, which BitmapPostings reports + // as absent — the same clean miss a real store returns. + out[i] = event.BitmapPostings(f.bitmaps[k]) } return out, nil } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_format.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_format.go index a1aa9f70e..ea5454ad0 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_format.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_format.go @@ -74,13 +74,57 @@ const ( // records. The Format value identifies the on-disk codec; readers // dispatch on it to select a matching RecordDecoder. eventsPackFormat packfile.Format = 0xFE1E000C // "Fellow Events 0xC" (zstd) - indexPackFormat packfile.Format = 0xFE1E000B // "Fellow Events 0xB" + // Bumped from 0xFE1E000B when an index.pack item body gained its + // leading codec byte. 0xFE1E000C was already taken by eventsPackFormat, + // so the index side skips to 0xD. An older reader pointed at a new + // artifact (or the reverse) fails at open rather than decoding a roaring + // header out of a delta body. + indexPackFormat packfile.Format = 0xFE1E000D // "Fellow Events 0xD" (codec byte) ) +// index.pack item codecs — the codec byte in the layout cold_index.go +// documents. +// +// The byte is what makes a future codec additive: a new one takes the next +// value, the reader learns to decode it, and artifacts written by either +// writer stay readable. Only a change to what the EXISTING values mean would +// need another indexPackFormat bump. +const ( + // itemCodecRoaring bodies are a serialized roaring bitmap, RunOptimize'd + // before Bitmap.WriteTo. + itemCodecRoaring byte = 0x00 + // itemCodecDelta bodies are AppendPostings output: a uvarint count + // then the event IDs, first absolute and the rest strictly-positive + // deltas. + itemCodecDelta byte = 0x01 +) + +// deltaPostingMaxCardinality is the term cardinality at or below which +// postings are stored as a delta-varint list instead of a roaring bitmap. +// +// Measured on a real pubnet chunk (10,000 ledgers, 8.65M events, 204,752 +// terms): below ~1024 postings roaring costs 1.7x to 3.8x a delta list, and +// above it roaring's run and bitmap containers pull ahead, reaching 12.5x on +// the nine terms holding more than a million postings. 99.2% of terms fall +// below the threshold and hold 71% of the byte saving. +// +// Cardinality is a proxy for whichever encoding is actually smaller, not a +// derivation of it: a contiguous 1024-posting term is far smaller as a run +// container than as deltas, and a scattered 262K-posting term is smaller as +// deltas than as roaring. The proxy holds on pubnet event postings because +// cardinality correlates with run structure there, fat terms being the +// clustered ones. That is a property of the data, so re-measure before +// retuning. +// +// This is WRITER POLICY, not format: the codec byte says what a record +// holds, so moving the threshold changes only what future artifacts choose +// and needs no re-backfill of the ones already written. +const deltaPostingMaxCardinality = 1024 + // IndexRecordFingerprintLen is the byte width of the leading // fingerprint in every index.pack record. The cold reader checks // this against the queried term's first four bytes to filter MPHF -// false positives before deserializing the bitmap. +// false positives before decoding the postings. const IndexRecordFingerprintLen = 4 // ────────────────────────────────────────────────────────────────── diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index.go index f9c49acfb..9c3ac2804 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index.go @@ -9,7 +9,6 @@ package event // MPHF wrapper live in cold_format.go. import ( - "bytes" "context" "errors" "fmt" @@ -45,21 +44,13 @@ func ColdIndexSecret(catalogSecret []byte, chunkID chunk.ID) [stores.SecretLen]b // A zero-term bitmaps (an eventless chunk, e.g. a pre-Soroban // backfill range) produces a real (empty) index.hash over zero terms // plus a zero-record index.pack. The cold reader resolves every -// LookupKeys entry against it to a nil-bitmap miss through the -// ordinary path, so neither readers nor orchestrators need a +// LookupKeys entry against it to a clean miss through the ordinary +// path, so neither readers nor orchestrators need a // pack-without-index special case. // -// bitmaps is the complete term index for the Chunk, uniquely owned by -// the caller (no concurrent reader holds a pointer to any of its -// bitmaps). WriteColdIndex mutates each bitmap in place via -// RunOptimize before MarshalBinary — RunOptimize re-encodes long runs -// of set bits as RUN containers, which MarshalBinary then serializes -// more compactly. For chunk 5999 the on-disk shrink is ~14% (108MB → -// 93MB), concentrated in dense, clustered terms (popular contracts, -// common topic[0] verbs). This pairs with the fastaggregation.go fix -// in RoaringBitmap/roaring#81 — without that fix, the RUN containers -// hit a slow (*Bitmap).lazyOR path at query time and K≥12 regresses -// catastrophically. +// bitmaps is the complete term index for the Chunk. It is read, not +// modified: encodeIndexBody builds its own bitmap for the terms that +// need one. // // Both cold backfill and the live-chunk freeze build a Bitmaps single-threaded by // re-deriving terms from raw LCMs (per-event TermsForBytes + Bitmaps.AddTo) and @@ -72,12 +63,13 @@ func ColdIndexSecret(catalogSecret []byte, chunkID chunk.ID) [stores.SecretLen]b // // offset size field // 0 4 fingerprint (first 4 bytes of the TermKey hash) -// 4 N serialized roaring bitmap (Bitmap.MarshalBinary) +// 4 1 codec (itemCodecRoaring | itemCodecDelta) +// 5 N postings in that codec // // The cold reader uses mphf.Lookup(term) → slot to find the record // position, packfile.Reader.ReadItem(slot, ...) to read the bytes, // verifies the 4-byte fingerprint against term[:4], and then -// deserializes the bitmap on match. Unseen terms still produce a +// decodes the postings on match. Unseen terms still produce a // slot (vanilla MPHF semantics) but their fingerprint mismatches — // the cold reader rejects them at that point. // @@ -131,11 +123,11 @@ func WriteColdIndex( // App fingerprint stays on the ORIGINAL term key, not the routed key. var fp [IndexRecordFingerprintLen]byte copy(fp[:], term[:IndexRecordFingerprintLen]) - // Mutate in place — bitmaps is uniquely owned by the caller, built - // single-threaded either way: cold backfill from the .pack, or the freeze - // from the read-only hot DB. - bitmap.RunOptimize() - entries = append(entries, indexEntry{slot: slot, fp: fp, bitmap: bitmap}) + body, berr := encodeIndexBody(nil, bitmap.ToArray()) + if berr != nil { + return berr + } + entries = append(entries, indexEntry{slot: slot, fp: fp, body: body}) } sort.Slice(entries, func(i, j int) bool { return entries[i].slot < entries[j].slot }) @@ -153,12 +145,12 @@ func WriteColdIndex( // for the rationale: offset-array size is per-record, so larger // records shrink the resident array proportionally). // - // No record codec is used: roaring's MarshalBinary already - // container-encodes (array / bitmap / RLE) the underlying data, - // and a second compression pass slows the query hot path - // measurably (~3.6× lookup latency in measurement) for marginal - // byte savings. Contrast events.pack, where XDR payloads grouped - // at 128/record offer plenty of compression headroom. + // No record codec is used: an item body is already + // container-encoded (roaring's array / bitmap / RLE) or + // delta-varint, and a second compression pass slows the query hot + // path measurably (~3.6× lookup latency in measurement) for + // marginal byte savings. Contrast events.pack, where XDR payloads + // grouped at 128/record offer plenty of compression headroom. pw, err := packfile.Create(indexPackPath, packfile.WriterOptions{ Format: indexPackFormat, ItemsPerRecord: indexPackItemsPerRecord, @@ -182,27 +174,54 @@ func WriteColdIndex( } // indexEntry is one assembled index.pack record: the slot it lands at, -// the 4-byte fingerprint, and the bitmap to serialize. +// the 4-byte fingerprint, and the encoded item body (codec ‖ postings). type indexEntry struct { - slot uint32 - fp [IndexRecordFingerprintLen]byte - bitmap *roaring.Bitmap + slot uint32 + fp [IndexRecordFingerprintLen]byte + body []byte +} + +// encodeIndexBody appends the index.pack item body for ids — the codec byte, +// then the postings in that codec — to dst. ids must be ascending and deduped. +// +// It is the single encoder every index.pack builder goes through, so no two +// of them can disagree about a term's encoding. It takes ids rather than a +// bitmap because the delta codec needs a sorted slice, while a bitmap is +// cheap to build from ids for the terms that need one. The one term's array +// that costs on the way in is transient and bounded by that term's +// cardinality — noise in an offline build that already holds the whole +// Bitmaps in memory. +func encodeIndexBody(dst []byte, ids []uint32) ([]byte, error) { + if len(ids) == 0 { + // A zero-posting term has no reason to exist, and the delta codec + // cannot represent one: DecodePostings rejects a zero count, so + // writing it would produce a record no reader can decode. + return nil, errors.New("events: index body for a term with no postings") + } + if len(ids) <= deltaPostingMaxCardinality { + return AppendPostings(append(dst, itemCodecDelta), ids), nil + } + // RunOptimize re-encodes long runs of set bits as RUN containers, which + // the serialization then writes more compactly: ~14% on chunk 5999, + // concentrated in dense clustered terms (popular contracts, common + // topic[0] verbs). It is also what makes the serialization canonical. + // This pairs with the fastaggregation.go fix in RoaringBitmap/roaring#81 + // — without that fix, the RUN containers hit a slow (*Bitmap).lazyOR + // path at query time and K≥12 regresses catastrophically. + bm := roaring.BitmapOf(ids...) + bm.RunOptimize() + b, err := bm.ToBytes() + if err != nil { + return nil, fmt.Errorf("events: serialize bitmap: %w", err) + } + return append(append(dst, itemCodecRoaring), b...), nil } // writeIndexPackEntries appends every assembled record to the index.pack // writer in slot order and finishes the pack. func writeIndexPackEntries(pw *packfile.Writer, entries []indexEntry) error { - // Serialize each bitmap into one reused buffer rather than a fresh - // MarshalBinary slice per record. AppendItem copies its input, so the - // buffer is safe to reuse across iterations; roaring's WriteTo emits - // the same bytes MarshalBinary would, so the pack is byte-identical. - var buf bytes.Buffer for _, e := range entries { - buf.Reset() - if _, werr := e.bitmap.WriteTo(&buf); werr != nil { - return fmt.Errorf("events: serialize bitmap at slot %d: %w", e.slot, werr) - } - if err := pw.AppendItem(e.fp[:], buf.Bytes()); err != nil { + if err := pw.AppendItem(e.fp[:], e.body); err != nil { return fmt.Errorf("events: write slot %d to index.pack: %w", e.slot, err) } } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index_test.go index a0d5eee0f..b26f21f0d 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_index_test.go @@ -9,7 +9,6 @@ import ( "path/filepath" "testing" - "github.com/RoaringBitmap/roaring/v2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -122,18 +121,15 @@ func TestWriteIndex_RoundTripsBitmapsPerTerm(t *testing.T) { record, ok := records[int(slot)] require.True(t, ok, "record missing at slot %d (term-%d)", slot, i) - require.GreaterOrEqual(t, len(record), IndexRecordFingerprintLen, "record at slot %d too short", slot) - - // Fingerprint must match term[:4]. - assert.Equal(t, term[:IndexRecordFingerprintLen], record[:IndexRecordFingerprintLen], - "fingerprint mismatch at slot %d", slot) - - // Deserialize bitmap. - bm := roaring.New() - require.NoError(t, bm.UnmarshalBinary(record[IndexRecordFingerprintLen:])) - assert.Equal(t, uint64(2), bm.GetCardinality(), "term-%d bitmap card", i) - assert.True(t, bm.Contains(uint32(i*10)), "term-%d missing event id %d", i, i*10) - assert.True(t, bm.Contains(uint32(i*10+1)), "term-%d missing event id %d", i, i*10+1) + + // Read it back the way the cold reader does, which also checks the + // fingerprint against term[:4] and dispatches on the codec byte. + post, derr := verifyAndDecodePostings(record, term, slot) + require.NoError(t, derr, "term-%d", i) + require.True(t, post.Present(), "term-%d must resolve, not read as a fingerprint miss", i) + assert.Equal(t, uint64(2), post.Cardinality(), "term-%d posting count", i) + assert.True(t, post.Contains(uint32(i*10)), "term-%d missing event id %d", i, i*10) + assert.True(t, post.Contains(uint32(i*10+1)), "term-%d missing event id %d", i, i*10+1) } } @@ -308,9 +304,9 @@ func TestWriteIndex_LargeIndex(t *testing.T) { } func TestWriteIndex_RecordEncoding(t *testing.T) { - // Lock the on-disk record format: fingerprint || roaring bitmap. - // Future readers (PR-3a) rely on this layout; if it ever changes - // silently, this test fails. + // Lock the on-disk record format: fingerprint || codec || postings. + // The cold reader relies on this layout; if it ever changes silently, + // this test fails. dir := t.TempDir() idx := NewBitmaps() idx.AddTo(ComputeTermKey([]byte("only"), FieldContractID), 42) @@ -326,13 +322,174 @@ func TestWriteIndex_RecordEncoding(t *testing.T) { term := ComputeTermKey([]byte("only"), FieldContractID) assert.Equal(t, term[:IndexRecordFingerprintLen], record[:IndexRecordFingerprintLen]) - bm := roaring.New() - require.NoError(t, bm.UnmarshalBinary(record[IndexRecordFingerprintLen:])) - assert.Equal(t, uint64(1), bm.GetCardinality()) - assert.True(t, bm.Contains(42)) + // A single-posting term takes the delta codec, and its body is the codec + // byte plus AppendPostings output — no roaring container framing. + assert.Equal(t, itemCodecDelta, record[IndexRecordFingerprintLen], + "a 1-posting term must use the delta codec") + post, derr := verifyAndDecodePostings(record, term, 0) + require.NoError(t, derr) + require.True(t, post.Present()) + assert.Equal(t, uint64(1), post.Cardinality()) + assert.True(t, post.Contains(42)) + assert.NotNil(t, post.IDs(), "a delta term must come back as an ID list") // Defensive: the fingerprint occupies bytes 0..3 in little-endian // the way TermKey itself encodes — read it back via binary helpers // just to lock the endianness contract. _ = binary.LittleEndian.Uint32(record[:IndexRecordFingerprintLen]) } + +// TestDeltaPostingThresholdPinned pins the threshold's value. Every other +// assertion names the constant symbolically, so they all hold at any value, +// including one that would silently cost 40% of the pack. +func TestDeltaPostingThresholdPinned(t *testing.T) { + require.Equal(t, 1024, deltaPostingMaxCardinality, + "retuning the threshold changes index.pack size materially; re-measure a real pubnet chunk first") +} + +// TestWriteIndex_CodecDispatch pins the cardinality dispatch at the threshold +// and round-trips both codecs through the reader's decode. +func TestWriteIndex_CodecDispatch(t *testing.T) { + dir := t.TempDir() + idx := NewBitmaps() + + term := func(name string, card int) TermKey { + k := ComputeTermKey([]byte(name), FieldContractID) + ids := make([]uint32, card) + for i := range ids { + ids[i] = uint32(i * 7) + } + idx.AddTo(k, ids...) + return k + } + single := term("single", 1) + atThreshold := term("at", deltaPostingMaxCardinality) + overThreshold := term("over", deltaPostingMaxCardinality+1) + + require.NoError(t, WriteColdIndex(context.Background(), indexTestChunkID, idx, dir, testIndexSecret)) + + m, err := openMPHF(filepath.Join(dir, IndexHashName(indexTestChunkID))) + require.NoError(t, err) + t.Cleanup(func() { _ = m.Close() }) + records := loadIndexPack(t, filepath.Join(dir, IndexPackName(indexTestChunkID))) + + codecOf := func(k TermKey) byte { + slot, lerr := m.Lookup(k) + require.NoError(t, lerr) + return records[int(slot)][IndexRecordFingerprintLen] + } + assert.Equal(t, itemCodecDelta, codecOf(atThreshold), + "exactly deltaPostingMaxCardinality postings must still be delta") + assert.Equal(t, itemCodecRoaring, codecOf(overThreshold), + "one posting past the threshold must be roaring") + + // Both codecs must come back through the reader's dispatch with the right + // postings. (TestColdIndex_BothCodecsOnDiskRoundTrip drives the same pair + // through the full ColdReader.) + wantCard := []uint64{1, deltaPostingMaxCardinality, deltaPostingMaxCardinality + 1} + for i, k := range []TermKey{single, atThreshold, overThreshold} { + want := wantCard[i] + slot, lerr := m.Lookup(k) + require.NoError(t, lerr) + post, derr := verifyAndDecodePostings(records[int(slot)], k, slot) + require.NoError(t, derr, "term %d", i) + require.True(t, post.Present(), "term %d must resolve", i) + assert.Equal(t, want, post.Cardinality(), "term %d cardinality", i) + assert.True(t, post.Contains(0), "term %d must hold its first posting", i) + assert.Equal(t, want <= deltaPostingMaxCardinality, post.IDs() != nil, + "term %d form must follow its codec", i) + } + + // An unknown codec byte must fail loudly rather than decode as something. + bad := append([]byte(nil), records[0]...) + bad[IndexRecordFingerprintLen] = 0x7f + var key TermKey + copy(key[:], bad[:IndexRecordFingerprintLen]) + _, derr := verifyAndDecodePostings(bad, key, 0) + require.ErrorContains(t, derr, "unknown codec") +} + +// TestEncodeIndexBodyRejectsEmpty pins that the encoder refuses a term with no +// postings rather than writing a record the reader cannot decode: the delta +// codec's count is a uvarint and DecodePostings rejects zero. +func TestEncodeIndexBodyRejectsEmpty(t *testing.T) { + _, err := encodeIndexBody(nil, nil) + require.Error(t, err) + + body, err := encodeIndexBody(nil, []uint32{7}) + require.NoError(t, err) + ids, err := DecodePostings(body[1:]) + require.NoError(t, err, "one posting must round-trip") + require.Equal(t, []uint32{7}, ids) +} + +// TestColdIndex_BothCodecsOnDiskRoundTrip is the format's own evidence. It +// builds a real cold artifact set through the production writer, reads the +// raw index.pack records back and proves BOTH codec bytes actually occur on +// disk, then resolves every term through a real ColdReader and checks the +// postings equal what went in. +// +// The corpus is shaped so the writer has to choose: one term holds every +// event (past the threshold, so roaring), and hundreds hold three each (well +// under it, so delta). +func TestColdIndex_BothCodecsOnDiskRoundTrip(t *testing.T) { + const chunkID = chunk.ID(0) + // One posting past the threshold is the smallest corpus that forces the + // roaring codec, so the test stays cheap while still crossing the line. + const fatCard = deltaPostingMaxCardinality + 1 + + dir := t.TempDir() + first := chunkID.FirstLedger() + + cw, err := NewColdWriter(chunkID, dir, ColdWriterOptions{}) + require.NoError(t, err) + t.Cleanup(func() { _ = cw.Close() }) + + idx := NewBitmaps() + offsets := NewLedgerOffsets(first) + want := make(map[TermKey][]uint32) + + fat := ComputeTermKey([]byte("fat"), FieldContractID) + for id := range uint32(fatCard) { + require.NoError(t, cw.Append(makeColdPayload(first, 1, fmt.Sprintf("e%d", id)))) + small := ComputeTermKey(fmt.Appendf(nil, "small-%d", id/3), FieldTopic0) + idx.AddTo(fat, id) + idx.AddTo(small, id) + want[fat] = append(want[fat], id) + want[small] = append(want[small], id) + } + require.NoError(t, offsets.Append(first, fatCard)) + require.NoError(t, cw.Finish(offsets)) + require.NoError(t, WriteColdIndex(context.Background(), chunkID, idx, dir, testIndexSecret)) + + // ── On-disk evidence ── + records := loadIndexPack(t, filepath.Join(dir, IndexPackName(chunkID))) + require.Len(t, records, len(want), "one index.pack record per term") + codecs := make(map[byte]int, 2) + for slot, record := range records { + require.Greater(t, len(record), IndexRecordFingerprintLen, "record at slot %d has no body", slot) + codecs[record[IndexRecordFingerprintLen]]++ + } + assert.Equal(t, 1, codecs[itemCodecRoaring], "exactly the fat term sits above the threshold") + assert.Equal(t, len(want)-1, codecs[itemCodecDelta], "every other term sits below it") + require.Len(t, codecs, 2, "both codecs, and only those two, must appear on disk") + + // ── Round trip through the real reader ── + cr, err := OpenColdReader(chunkID, dir, ColdReaderOptions{}) + require.NoError(t, err) + t.Cleanup(func() { _ = cr.Close() }) + + keys := make([]TermKey, 0, len(want)) + for k := range want { + keys = append(keys, k) + } + got, err := cr.LookupKeys(context.Background(), keys) + require.NoError(t, err) + require.Len(t, got, len(keys)) + for i, k := range keys { + require.True(t, got[i].Present(), "term %x must resolve", k) + assert.Equal(t, want[k], got[i].SelectIDs(0, false), "term %x postings", k) + assert.Equal(t, len(want[k]) <= deltaPostingMaxCardinality, got[i].IDs() != nil, + "term %x must come back in the form its codec stored", k) + } +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader.go index 8030dc20b..e8b1efcce 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader.go @@ -175,28 +175,16 @@ func OpenColdReader(chunkID chunk.ID, bucketDir string, opts ColdReaderOptions) if err != nil { return err } - if idx.isEmpty() { - // A zero-term index is only valid for an eventless chunk: cross-check - // events.pack's count so a mispaired empty index fails loudly instead - // of silently matching nothing. - m, merr := c.waitMeta() - if merr != nil { - return fmt.Errorf("events: validate empty index for chunk %s: %w", c.chunkID, merr) - } - if m.count != 0 { - return fmt.Errorf( - "events: %s holds zero terms but events.pack holds %d events for chunk %s (torn or mispaired index)", - indexHashPath, m.count, c.chunkID) - } - return nil - } - // Non-empty index: bind the pair to this chunk before serving from - // it — index.pack/index.hash carry no chunk ID of their own, so a + // Bind the index pair to this chunk before serving from it — + // index.pack/index.hash carry no chunk ID of their own, so a // mispaired index would silently return an incomplete subset of // matches. Three cheap checks: index.pack's trailer Format, // index.hash keys == index.pack records (halves of one build), and - // non-empty index ⇒ non-empty events.pack (converse of the - // empty-index check above). + // the empty/non-empty cross-checks against events.pack below. The + // trailer checks run for the eventless case too: a zero-term chunk + // still writes a real (zero-record) index.pack, so a stale-format + // or foreign pack fails here instead of hiding behind the + // empty-index early return. tr, terr := c.index.Trailer() if terr != nil { return fmt.Errorf("events: open %s: %w", indexPackPath, terr) @@ -211,6 +199,23 @@ func OpenColdReader(chunkID chunk.ID, bucketDir string, opts ColdReaderOptions) "but index.pack holds %d records (mispaired artifacts)", c.chunkID, idx.numKeys(), tr.TotalItems) } + if idx.isEmpty() { + // A zero-term index is only valid for an eventless chunk: cross-check + // events.pack's count so a mispaired empty index fails loudly instead + // of silently matching nothing. + m, merr := c.waitMeta() + if merr != nil { + return fmt.Errorf("events: validate empty index for chunk %s: %w", c.chunkID, merr) + } + if m.count != 0 { + return fmt.Errorf( + "events: %s holds zero terms but events.pack holds %d events for chunk %s (torn or mispaired index)", + indexHashPath, m.count, c.chunkID) + } + return nil + } + // Non-empty index ⇒ non-empty events.pack (converse of the + // empty-index check above). m, merr := c.waitMeta() if merr != nil { return fmt.Errorf("events: validate index for chunk %s: %w", c.chunkID, merr) @@ -303,35 +308,61 @@ func (c *ColdReader) Offsets() (*LedgerOffsets, error) { return m.offsets, nil } -// verifyAndDeserializeBitmap checks the index.pack record's leading -// fingerprint against key's prefix and, on match, unmarshals a fresh -// bitmap. On fingerprint mismatch (residual MPHF collision on an -// unseen key) it returns (nil, nil) — the caller treats nil as -// not-found. record is valid only inside ReadItem's callback; -// UnmarshalBinary copies into roaring's internal state so the -// returned bitmap outlives the callback safely. -func verifyAndDeserializeBitmap(record []byte, key TermKey, slot uint32) (*roaring.Bitmap, error) { - if len(record) < IndexRecordFingerprintLen { - return nil, fmt.Errorf("events: index.pack record at slot %d truncated (%d bytes)", slot, len(record)) +// verifyAndDecodePostings checks the index.pack record's leading fingerprint +// against key's prefix and, on match, decodes its postings. On fingerprint +// mismatch (residual MPHF collision on an unseen key) it returns the zero +// Postings, which the caller treats as not-found. +// +// A delta term is returned un-materialized (see Intersect). record is +// valid only inside ReadItems' callback, and neither codec aliases it, so the +// result outlives the callback safely. +func verifyAndDecodePostings(record []byte, key TermKey, slot uint32) (Postings, error) { + if len(record) <= IndexRecordFingerprintLen { + return Postings{}, fmt.Errorf( + "events: index.pack record at slot %d truncated (%d bytes)", slot, len(record)) } if !bytes.Equal(record[:IndexRecordFingerprintLen], key[:IndexRecordFingerprintLen]) { - return nil, nil //nolint:nilnil // not-found signaled by nil bitmap, no error - } - bm := roaring.New() - if err := bm.UnmarshalBinary(record[IndexRecordFingerprintLen:]); err != nil { - return nil, fmt.Errorf("events: unmarshal bitmap at slot %d: %w", slot, err) + return Postings{}, nil + } + body := record[IndexRecordFingerprintLen:] + switch body[0] { + case itemCodecRoaring: + bm := roaring.New() + if err := bm.UnmarshalBinary(body[1:]); err != nil { + return Postings{}, fmt.Errorf("events: unmarshal bitmap at slot %d: %w", slot, err) + } + // Byte integrity is the packfile layer's job (#910 checksums + // index.pack record payloads), so the decode trusts what it is + // handed: no structural validation here. bm.Validate() in + // particular is banned from this path — its run-disjointness proof + // is super-linear on run-dense bitmaps and measured 15ms → 230ms + // p99 on cold pubnet events queries. A parseable-but-hollow body + // decodes to zero postings: a genuinely empty one normalizes to a + // clean miss in BitmapPostings, and a zero-interval run container + // (whose container survives IsEmpty) reads as a present term whose + // intersections and unions are simply empty — the trust level the + // base read path always had. + return BitmapPostings(bm), nil + case itemCodecDelta: + ids, err := DecodePostings(body[1:]) + if err != nil { + return Postings{}, fmt.Errorf("events: decode postings at slot %d: %w", slot, err) + } + return IDPostings(ids), nil + default: + return Postings{}, fmt.Errorf( + "events: index.pack record at slot %d has unknown codec 0x%02x", slot, body[0]) } - return bm, nil } -// LookupKeys returns bitmaps for each key, aligned positionally with -// the input slice (result[i] corresponds to keys[i]). See -// Reader.LookupKeys for the semantics. +// LookupKeys returns each key's postings, aligned positionally with the input +// slice (result[i] corresponds to keys[i]). See Reader.LookupKeys for the +// semantics. // // Cold-side implementation: // // 1. MPHF-resolve every key. Keys rejected at the routing stage -// (streamhash ErrKeyNotFound) get result[i] = nil and never +// (streamhash ErrKeyNotFound) leave result[i] absent and never // touch index.pack. // 2. Sort the surviving (key, slot) pairs by slot and dedupe — // pathological residual collisions can map two distinct keys @@ -341,11 +372,11 @@ func verifyAndDeserializeBitmap(record []byte, key TermKey, slot uint32) (*roari // calls and fans out across the worker count configured via // ColdReaderOptions.Concurrency. // 4. In the callback, verify each pending key's fingerprint -// against the record header and unmarshal a fresh bitmap per -// match. Misses (fingerprint mismatch) leave result[i] = nil. +// against the record header and decode its postings per match. +// Misses (fingerprint mismatch) leave result[i] absent. // //nolint:cyclop // the four documented steps above, inline; splitting obscures the pass structure -func (c *ColdReader) LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring.Bitmap, error) { +func (c *ColdReader) LookupKeys(ctx context.Context, keys []TermKey) ([]Postings, error) { if c.closed.Load() { return nil, stores.ErrStoreClosed } @@ -364,7 +395,7 @@ func (c *ColdReader) LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring return nil, err } - results := make([]*roaring.Bitmap, len(keys)) + results := make([]Postings, len(keys)) type pendingKey struct { outIdx int @@ -375,7 +406,7 @@ func (c *ColdReader) LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring slot, err := mphf.Lookup(key) if err != nil { if errors.Is(err, ErrKeyNotFound) { - continue // result[i] stays nil + continue // result[i] stays absent } return nil, fmt.Errorf("events: LookupKeys MPHF for chunk %s: %w", c.chunkID, err) } @@ -403,16 +434,15 @@ func (c *ColdReader) LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring if err := c.index.ReadItems(ctx, positions, func(readIdx int, record []byte) error { // Multiple pending keys may share this slot (residual MPHF - // collision). verifyAndDeserializeBitmap returns a fresh - // bitmap per match and (nil, nil) on fingerprint mismatch — - // leaving results[outIdx] = nil for misses. + // collision). verifyAndDecodePostings decodes per match and returns + // the zero Postings on fingerprint mismatch, leaving that key absent. for _, pIdx := range pendingBySlot[readIdx] { p := pending[pIdx] - bm, err := verifyAndDeserializeBitmap(record, keys[p.outIdx], p.slot) + post, err := verifyAndDecodePostings(record, keys[p.outIdx], p.slot) if err != nil { return err } - results[p.outIdx] = bm + results[p.outIdx] = post } return nil }); err != nil { diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader_test.go index 0b135237e..84cf8f59e 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/cold_reader_test.go @@ -5,8 +5,10 @@ import ( "fmt" "os" "path/filepath" + "slices" "testing" + "github.com/RoaringBitmap/roaring/v2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -158,30 +160,6 @@ func TestColdReader_LookupUnseenTermReturnsNil(t *testing.T) { } } -func TestColdReader_LookupKeysReturnsFreshBitmaps(t *testing.T) { - // Two LookupKeys calls for the same term return independent - // bitmaps — mutating one must not affect the other. The interface - // contract says cold-side results are freshly unmarshaled and - // logically owned by the caller; this pins that a mutation cannot - // bleed into a later lookup. - const chunkID = chunk.ID(0) - dir, payloads := buildColdFixture(t, chunkID, 8, 1) - - cr, err := OpenColdReader(chunkID, dir, ColdReaderOptions{}) - require.NoError(t, err) - t.Cleanup(func() { _ = cr.Close() }) - - key := contractTermKey(payloads[0]) - first := lookupOne(t, cr, key) - require.NotNil(t, first) - first.Add(999_999) - - second := lookupOne(t, cr, key) - require.NotNil(t, second) - assert.False(t, second.Contains(999_999), - "a LookupKeys result mutation must not bleed into the next lookup") -} - func TestColdReader_FetchEventsRoundTrip(t *testing.T) { const chunkID = chunk.ID(0) dir, payloads := buildColdFixture(t, chunkID, 5, 1) @@ -273,13 +251,13 @@ func TestColdReader_EventlessChunk(t *testing.T) { require.NoError(t, err) assert.Zero(t, cnt) - // Term-filtered paths miss cleanly (nil bitmap, no error) instead - // of surfacing a filesystem error. + // Term-filtered paths miss cleanly (absent postings, no error) + // instead of surfacing a filesystem error. someTerm := ComputeTermKey([]byte("any"), FieldContractID) - bms, err := cr.LookupKeys(context.Background(), []TermKey{someTerm}) + got, err := cr.LookupKeys(context.Background(), []TermKey{someTerm}) require.NoError(t, err) - require.Len(t, bms, 1) - assert.Nil(t, bms[0]) + require.Len(t, got, 1) + assert.False(t, got[0].Present()) // Full-scan path yields nothing. for _, err := range cr.All(context.Background()) { @@ -297,7 +275,9 @@ func TestColdReader_EventlessChunk(t *testing.T) { // guarding a zero-term index: it is only valid when events.pack is also // empty. A real empty index.hash mispaired onto a chunk whose events.pack // holds events must fail loudly at the first lookup, not silently match -// nothing. +// nothing. An empty index.hash over the chunk's populated index.pack trips +// the pairing check; mispairing BOTH empty halves agrees with pairing and +// format, so the events.pack cross-check is the layer that catches it. func TestColdReader_EmptyIndexOverNonEmptyPackErrors(t *testing.T) { const chunkID = chunk.ID(0) dir, payloads := buildColdFixture(t, chunkID, 2, 1) @@ -317,6 +297,20 @@ func TestColdReader_EmptyIndexOverNonEmptyPackErrors(t *testing.T) { // The mismatch must surface as an error, not a clean miss (nil, no error). _, lerr := cr.LookupKeys(context.Background(), []TermKey{contractTermKey(payloads[0])}) require.Error(t, lerr, "a mispaired empty index must error, not miss silently (nil bitmap)") + assert.Contains(t, lerr.Error(), "index pair mismatch", + "empty index.hash over the populated index.pack is the pairing check's catch") + + // Mispair BOTH empty halves onto the non-empty chunk: pairing (0 == 0) + // and format agree, so the events.pack cross-check must catch it. + emptyPack, err := os.ReadFile(filepath.Join(emptyDir, IndexPackName(chunkID))) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(dir, IndexPackName(chunkID)), emptyPack, 0o644)) + + cr2, err := OpenColdReader(chunkID, dir, ColdReaderOptions{}) + require.NoError(t, err) + t.Cleanup(func() { _ = cr2.Close() }) + _, lerr = cr2.LookupKeys(context.Background(), []TermKey{contractTermKey(payloads[0])}) + require.Error(t, lerr, "both-empty halves over a non-empty chunk must still error") assert.Contains(t, lerr.Error(), "holds zero terms") } @@ -548,26 +542,28 @@ func TestColdReader_LookupKeys(t *testing.T) { missing := ComputeTermKey([]byte("never-added"), FieldTopic1) keys := []TermKey{contractKey, missing, topicKey, missing} - bms, err := cr.LookupKeys(context.Background(), keys) + got, err := cr.LookupKeys(context.Background(), keys) require.NoError(t, err) - require.Len(t, bms, len(keys)) + require.Len(t, got, len(keys)) - require.NotNil(t, bms[0], "contract key should hit") - assert.Equal(t, uint64(len(payloads)), bms[0].GetCardinality()) + require.True(t, got[0].Present(), "contract key should hit") + assert.Equal(t, uint64(len(payloads)), got[0].Cardinality()) - assert.Nil(t, bms[1], "missing key should be nil") + assert.False(t, got[1].Present(), "missing key should be absent") - require.NotNil(t, bms[2], "topic key should hit") - assert.Equal(t, uint64(1), bms[2].GetCardinality()) - assert.True(t, bms[2].Contains(0)) + require.True(t, got[2].Present(), "topic key should hit") + assert.Equal(t, uint64(1), got[2].Cardinality()) + assert.True(t, got[2].Contains(0)) - assert.Nil(t, bms[3], "second missing must also be nil") + assert.False(t, got[3].Present(), "second missing must also be absent") } -// TestColdReader_LookupKeysClonesAcrossCalls pins that two -// LookupKeys batches for the same key return independent bitmaps — -// mutating one must not bleed into the next. -func TestColdReader_LookupKeysClonesAcrossCalls(t *testing.T) { +// TestColdReader_LookupKeysIsolatesCalls pins that two LookupKeys batches for +// the same key return independent postings, so scribbling on one cannot bleed +// into the next. The fixture's term is small, so it comes back as an ID slice; +// mutating the slice is the mutation that could actually reach a shared +// buffer, whereas Bitmap() on an ID-backed term hands back a throwaway. +func TestColdReader_LookupKeysIsolatesCalls(t *testing.T) { const chunkID = chunk.ID(0) dir, payloads := buildColdFixture(t, chunkID, 8, 1) @@ -579,14 +575,15 @@ func TestColdReader_LookupKeysClonesAcrossCalls(t *testing.T) { first, err := cr.LookupKeys(context.Background(), []TermKey{key}) require.NoError(t, err) require.Len(t, first, 1) - require.NotNil(t, first[0]) - first[0].Add(999_999) + require.NotNil(t, first[0].IDs(), "a small term must come back as an ID slice") + + want := slices.Clone(first[0].IDs()) + first[0].IDs()[0] = 999_999 second, err := cr.LookupKeys(context.Background(), []TermKey{key}) require.NoError(t, err) require.Len(t, second, 1) - require.NotNil(t, second[0]) - assert.False(t, second[0].Contains(999_999), + assert.Equal(t, want, second[0].IDs(), "a LookupKeys result mutation must not bleed into the next LookupKeys") } @@ -788,3 +785,46 @@ func TestColdReader_RejectsNonEmptyIndexOnEventlessChunk(t *testing.T) { _, err = cr.LookupKeys(context.Background(), []TermKey{contractTermKey(payloads[0])}) require.ErrorContains(t, err, "eventless") } + +// TestColdReader_TrustsChecksummedBytes pins the decode path's trust model: +// byte integrity belongs to the packfile layer (#910 checksums index.pack +// record payloads), so parseable-but-hollow roaring bodies are not treated as +// errors here — and, deliberately, no structural validation runs per decode +// (roaring's Validate is super-linear on run-dense bitmaps: measured +// 15ms → 230ms p99 on cold pubnet events queries). A genuinely empty body +// resolves to an absent term; a zero-interval run container — a shape +// UnmarshalBinary accepts — resolves to a present term with zero postings, +// which matches nothing. +func TestColdReader_TrustsChecksummedBytes(t *testing.T) { + empty, err := roaring.New().ToBytes() + require.NoError(t, err) + key := ComputeTermKey([]byte("hollow"), FieldContractID) + rec := append([]byte(nil), key[:IndexRecordFingerprintLen]...) + rec = append(rec, itemCodecRoaring) + rec = append(rec, empty...) + post, derr := verifyAndDecodePostings(rec, key, 7) + require.NoError(t, derr) + assert.False(t, post.Present(), "an empty body normalizes to an absent term") + + zeroRuns := []byte{ + 0x3B, 0x30, // SERIAL_COOKIE (12347) — run-container format + 0x00, 0x00, // size-1 = 0 → one container + 0x01, // run-flag bitset: container 0 is a run container + 0x00, 0x00, // key 0 + 0x00, 0x00, // cardinality-1 header claim + 0x00, 0x00, // numRuns = 0 + } + var bm roaring.Bitmap + require.NoError(t, bm.UnmarshalBinary(zeroRuns)) + require.False(t, bm.IsEmpty(), "the container exists, so IsEmpty misses it") + require.Zero(t, bm.GetCardinality()) + + rec2 := append([]byte(nil), key[:IndexRecordFingerprintLen]...) + rec2 = append(rec2, itemCodecRoaring) + rec2 = append(rec2, zeroRuns...) + post2, derr2 := verifyAndDecodePostings(rec2, key, 8) + require.NoError(t, derr2) + assert.True(t, post2.Present()) + assert.Zero(t, post2.Cardinality(), "a hollow container reads as a present term with zero postings") + assert.Empty(t, post2.SelectIDs(0, false)) +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go index dc6c44d08..36d89ed3f 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps.go @@ -81,11 +81,16 @@ func NewConcurrentBitmapsFromBitmaps(b Bitmaps) *ConcurrentBitmaps { return cb } -// Get returns the bitmap for the given term key, or (nil, nil) when -// the key is not indexed. The returned bitmap is an immutable -// snapshot: writers publish new termStates via atomic.Store, so the -// pointer this method returns will never be mutated by anyone — but -// only if callers respect the "read-only" half of the contract. +// Get returns the postings for the given term key, or the zero Postings when +// the key is not indexed. What comes back is an immutable snapshot: writers +// publish new termStates via atomic.Store, and neither a published bitmap nor +// a published id slice is ever written afterwards — but only if callers +// respect the "read-only" half of the contract. +// +// A sparse entry is returned as its id slice, un-materialized, since that is +// the form it is stored in and the form Intersect can drive from. +// Building a bitmap for it here was pure loss: the state already held an id +// list, and every Get paid for a copy the planner then took apart again. // // Forbidden caller-side methods on the returned bitmap (these have // side effects on the bitmap's internal needCopyOnWrite[] array, @@ -98,6 +103,8 @@ func NewConcurrentBitmapsFromBitmaps(b Bitmaps) *ConcurrentBitmaps { // - SetCopyOnWrite // - Any *Writable* accessor on the underlying roaringArray // +// The same rule applies to a returned id slice: read it, never write to it. +// // Safe caller-side methods (used by event.Matches today): any // non-mutating read — Contains, GetCardinality, Iterator, // ToArray, IsEmpty, Minimum, Maximum — plus the non-mutating @@ -107,8 +114,8 @@ func NewConcurrentBitmapsFromBitmaps(b Bitmaps) *ConcurrentBitmaps { // ≥2-input qualifier on FastAnd/FastOr: with a single input the // roaring library has historically taken a Clone-the-input // shortcut, so callers MUST avoid passing a singleton slice to -// those aggregators (the event store's Matches guards its single-input -// cases before calling FastAnd/FastOr). +// those aggregators (Intersect and Union guard their +// single-input cases before calling FastAnd/FastOr). // // Callers may hold the pointer arbitrarily long. A subsequent Get // on the same key may return either this same pointer (no AddTo @@ -118,22 +125,23 @@ func NewConcurrentBitmapsFromBitmaps(b Bitmaps) *ConcurrentBitmaps { // Concurrency: the RLock is held only for the map lookup. Once the // per-entry pointer is captured, the lock is released; the atomic // load on the entry happens lock-free. -func (s *ConcurrentBitmaps) Get(key TermKey) (*roaring.Bitmap, error) { +func (s *ConcurrentBitmaps) Get(key TermKey) (Postings, 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 + return Postings{}, nil } // The pointer is always Stored with a non-nil termState holding a // non-nil bm or non-empty ids before it is published to the map. st := p.Load() if st.bm != nil { - return st.bm, nil + return BitmapPostings(st.bm), nil } - bm := roaring.New() - bm.AddMany(st.ids) - return bm, nil + // Sparse: hand out the published id list. AddTo builds a fresh slice per + // publish and never appends into one already published, so holding it is + // safe for as long as the caller likes. + return IDPostings(st.ids), nil } // AddTo records each eventID under key. Idempotent: callers 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 index fc31d6e8e..84d78a405 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/concurrent_bitmaps_test.go @@ -23,21 +23,21 @@ func TestConcurrentBitmaps_AddToAndGet(t *testing.T) { s.AddTo(key, 1) s.AddTo(key, 2) - bm, err := s.Get(key) + post, 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)) + require.True(t, post.Present()) + assert.Equal(t, uint64(3), post.Cardinality()) + assert.True(t, post.Contains(0)) + assert.True(t, post.Contains(1)) + assert.True(t, post.Contains(2)) } func TestConcurrentBitmaps_GetMissing(t *testing.T) { s := newTestConcurrentBitmaps() key := ComputeTermKey([]byte("missing"), FieldTopic0) - bm, err := s.Get(key) + post, err := s.Get(key) require.NoError(t, err) - assert.Nil(t, bm) + assert.False(t, post.Present()) } func TestConcurrentBitmaps_ListMode(t *testing.T) { @@ -55,10 +55,15 @@ func TestConcurrentBitmaps_ListMode(t *testing.T) { assert.Nil(t, st.bm, "must still be in list mode") assert.Len(t, st.ids, promotionThreshold-1) - bm, err := s.Get(key) + post, err := s.Get(key) require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(promotionThreshold-1), bm.GetCardinality()) + require.True(t, post.Present()) + assert.Equal(t, uint64(promotionThreshold-1), post.Cardinality()) + + // The form follows storage: a sparse term is never materialized, and what + // comes back is the store's own slice rather than a copy. + require.NotNil(t, post.IDs(), "a sparse term must not be materialized") + assert.Same(t, &st.ids[0], &post.IDs()[0]) // Get must not have promoted. assert.Nil(t, p.Load().bm) @@ -78,6 +83,11 @@ func TestConcurrentBitmaps_Promotion(t *testing.T) { require.NotNil(t, st.bm) assert.Empty(t, st.ids, "sparse ids cleared after promotion") assert.Equal(t, uint64(promotionThreshold), st.bm.GetCardinality()) + + post, err := s.Get(key) + require.NoError(t, err) + require.True(t, post.Present()) + assert.Nil(t, post.IDs(), "a promoted term comes back as a bitmap") } func TestConcurrentBitmaps_AddAfterPromotion(t *testing.T) { @@ -90,11 +100,11 @@ func TestConcurrentBitmaps_AddAfterPromotion(t *testing.T) { s.AddTo(key, 1000) s.AddTo(key, 2000) - bm, err := s.Get(key) + post, 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)) + assert.Equal(t, uint64(promotionThreshold+2), post.Cardinality()) + assert.True(t, post.Contains(1000)) + assert.True(t, post.Contains(2000)) } func TestConcurrentBitmaps_BatchAddTo(t *testing.T) { @@ -103,12 +113,12 @@ func TestConcurrentBitmaps_BatchAddTo(t *testing.T) { s.AddTo(key, 0, 1, 2, 3, 4) - bm, err := s.Get(key) + post, 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)) + require.True(t, post.Present()) + assert.Equal(t, uint64(5), post.Cardinality()) + assert.True(t, post.Contains(0)) + assert.True(t, post.Contains(4)) } func TestConcurrentBitmaps_BatchAddToPromotion(t *testing.T) { @@ -146,13 +156,13 @@ func TestConcurrentBitmaps_GetReturnsImmutableSnapshot(t *testing.T) { before, err := s.Get(key) require.NoError(t, err) - beforeCard := before.GetCardinality() + beforeCard := before.Cardinality() // 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(), + assert.Equal(t, beforeCard, before.Cardinality(), "AddTo published a new snapshot; the borrowed pointer must remain unchanged") after, err := s.Get(key) @@ -189,10 +199,10 @@ func TestConcurrentBitmaps_ConcurrentGetIsSafe(t *testing.T) { wg.Go(func() { for range 50 { for _, k := range keys { - bm, err := s.Get(k) + post, err := s.Get(k) require.NoError(t, err) - require.NotNil(t, bm) - _ = bm.Contains(0) + require.True(t, post.Present()) + _ = post.Contains(0) } } }) @@ -201,9 +211,9 @@ func TestConcurrentBitmaps_ConcurrentGetIsSafe(t *testing.T) { // Sanity check: the store is untouched by all that reading. for _, k := range keys { - bm, err := s.Get(k) + post, err := s.Get(k) require.NoError(t, err) - require.NotNil(t, bm) + require.True(t, post.Present()) } } @@ -243,14 +253,14 @@ func TestConcurrentBitmaps_ConcurrentReadWrite(t *testing.T) { wg.Wait() for _, key := range keys { - bm, err := s.Get(key) + post, err := s.Get(key) require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(numEvents/numTerms), bm.GetCardinality()) + require.True(t, post.Present()) + assert.Equal(t, uint64(numEvents/numTerms), post.Cardinality()) } } -// TestConcurrentBitmaps_GetDuringPromotionNeverReturnsNil pins +// TestConcurrentBitmaps_GetDuringPromotionNeverReturnsAbsent 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 @@ -259,9 +269,9 @@ func TestConcurrentBitmaps_ConcurrentReadWrite(t *testing.T) { // 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 +// promotion boundary should never produce an absent result and // should never trip the race detector. -func TestConcurrentBitmaps_GetDuringPromotionNeverReturnsNil(t *testing.T) { +func TestConcurrentBitmaps_GetDuringPromotionNeverReturnsAbsent(t *testing.T) { s := newTestConcurrentBitmaps() const numKeys = 200 @@ -303,12 +313,12 @@ func TestConcurrentBitmaps_GetDuringPromotionNeverReturnsNil(t *testing.T) { default: } for _, k := range keys { - bm, err := s.Get(k) + post, 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") + // always observe a present term. + require.True(t, post.Present(), "Get returned absent during promotion window") } } }) @@ -340,12 +350,12 @@ func TestConcurrentBitmaps_AddToIsIdempotent(t *testing.T) { // And add a new one — must still go through. s.AddTo(key, 3) - bm, err := s.Get(key) + post, err := s.Get(key) require.NoError(t, err) - require.NotNil(t, bm) - assert.Equal(t, uint64(4), bm.GetCardinality()) + require.True(t, post.Present()) + assert.Equal(t, uint64(4), post.Cardinality()) for _, id := range []uint32{0, 1, 2, 3} { - assert.True(t, bm.Contains(id)) + assert.True(t, post.Contains(id)) } }) @@ -365,9 +375,9 @@ func TestConcurrentBitmaps_AddToIsIdempotent(t *testing.T) { s.AddTo(key, i) } - bm, err := s.Get(key) + post, err := s.Get(key) require.NoError(t, err) - assert.Equal(t, uint64(promotionThreshold), bm.GetCardinality()) + assert.Equal(t, uint64(promotionThreshold), post.Cardinality()) }) } @@ -439,22 +449,22 @@ func TestNewConcurrentBitmapsFromBitmaps_DirectlyPinsContract(t *testing.T) { cb := NewConcurrentBitmapsFromBitmaps(src) - bmA, err := cb.Get(keyA) + postA, err := cb.Get(keyA) require.NoError(t, err) - require.NotNil(t, bmA) - assert.Equal(t, uint64(5), bmA.GetCardinality()) - assert.True(t, bmA.GetCopyOnWrite(), + require.True(t, postA.Present()) + assert.Equal(t, uint64(5), postA.Cardinality()) + assert.True(t, postA.Bitmap().GetCopyOnWrite(), "FromBitmaps must enable CopyOnWrite so the live-ingest AddTo path Clones via the fast shallow path") - bmB, err := cb.Get(keyB) + postB, err := cb.Get(keyB) require.NoError(t, err) - require.NotNil(t, bmB) - assert.Equal(t, uint64(2), bmB.GetCardinality()) - assert.True(t, bmB.GetCopyOnWrite()) + require.True(t, postB.Present()) + assert.Equal(t, uint64(2), postB.Cardinality()) + assert.True(t, postB.Bitmap().GetCopyOnWrite()) - bmNil, err := cb.Get(keyNil) + postNil, err := cb.Get(keyNil) require.NoError(t, err) - assert.Nil(t, bmNil, "nil source entries must be skipped, not panicked") + assert.False(t, postNil.Present(), "nil source entries must be skipped, not panicked") // Subsequent AddTo on the converted index produces a new // termState whose bitmap still has CopyOnWrite (inherited via 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 f30fa809d..74fdaf2fb 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/hot_store.go @@ -8,7 +8,6 @@ import ( "iter" "math" - "github.com/RoaringBitmap/roaring/v2" "github.com/linxGnu/grocksdb" "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/chunk" @@ -168,16 +167,17 @@ 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). +// LookupKeys returns each key's postings, aligned positionally with +// the input slice. result[i] is the zero Postings if keys[i] +// has no matching events. See Reader.LookupKeys for the semantics — +// in particular the borrowed-result 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. -func (h *HotStore) LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring.Bitmap, error) { +func (h *HotStore) LookupKeys(ctx context.Context, keys []TermKey) ([]Postings, error) { if h.chunkStore.IsClosed() { return nil, stores.ErrStoreClosed } @@ -187,13 +187,13 @@ func (h *HotStore) LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring.B if len(keys) == 0 { return nil, nil } - results := make([]*roaring.Bitmap, len(keys)) + results := make([]Postings, len(keys)) for i, key := range keys { - bm, err := h.mirror.Get(key) + p, err := h.mirror.Get(key) if err != nil { return nil, 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[i] = p // absent for misses — Get already returns the zero Postings } return results, nil } 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 2a2a2180d..6df1a5644 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 @@ -399,9 +399,9 @@ func TestHotStore_PostCloseReadsError(t *testing.T) { require.NoError(t, ingestLedgerEvents(h.store, chunkID.FirstLedger(), []Payload{p})) require.NoError(t, h.raw.Close()) - // LookupKeys must error rather than silently returning cached bitmaps. - bms, err := h.store.LookupKeys(context.Background(), []TermKey{keys[0]}) - assert.Nil(t, bms) + // LookupKeys must error rather than silently returning cached postings. + post, err := h.store.LookupKeys(context.Background(), []TermKey{keys[0]}) + assert.Nil(t, post) require.ErrorIs(t, err, stores.ErrStoreClosed) // FetchEvents returns ErrStoreClosed. @@ -610,12 +610,18 @@ func firstIterError(seq iter.Seq2[Payload, error]) error { // 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. +// +// A bitmap-backed result comes back as the store's own bitmap, so the tests +// pinning the mirror's immutable-snapshot contract still see it. func lookupOne(t *testing.T, r Reader, key TermKey) *roaring.Bitmap { t.Helper() - bms, err := r.LookupKeys(context.Background(), []TermKey{key}) + got, err := r.LookupKeys(context.Background(), []TermKey{key}) require.NoError(t, err) - require.Len(t, bms, 1) - return bms[0] + require.Len(t, got, 1) + if !got[0].Present() { + return nil + } + return got[0].Bitmap() } func TestHotStore_FetchRangeMidRange(t *testing.T) { 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..891e324c1 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 @@ -22,10 +22,10 @@ func TestWarmup_FreshChunkProducesEmptyMirrorsViaNewWithStore(t *testing.T) { 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)) + // (absent postings, no error). + post, err := h.store.mirror.Get(ComputeTermKey([]byte("any"), FieldContractID)) require.NoError(t, err) - assert.Nil(t, bm) + assert.False(t, post.Present()) 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()) @@ -58,10 +58,10 @@ func TestWarmup_RebuildsMirrorFromIngestedRows(t *testing.T) { } expected := make(map[TermKey]uint64) for _, k := range seededKeys { - bm, err := hot1.mirror.Get(k) + post, 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.True(t, post.Present(), "seeded term missing from the pre-close mirror") + expected[k] = post.Cardinality() } require.NoError(t, raw1.Close()) @@ -70,10 +70,10 @@ func TestWarmup_RebuildsMirrorFromIngestedRows(t *testing.T) { got := make(map[TermKey]uint64) for k := range expected { - bm, err := hot2.mirror.Get(k) + post, err := hot2.mirror.Get(k) require.NoError(t, err) - require.NotNil(t, bm, "seeded term missing from the reopened mirror") - got[k] = bm.GetCardinality() + require.True(t, post.Present(), "seeded term missing from the reopened mirror") + got[k] = post.Cardinality() } assert.Equal(t, expected, got) } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/intersect.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/intersect.go new file mode 100644 index 000000000..71854694b --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/intersect.go @@ -0,0 +1,177 @@ +package event + +import ( + "cmp" + "slices" + + "github.com/RoaringBitmap/roaring/v2" +) + +// mergeRatio is the length ratio at which probing a sorted ID list switches +// from a cursored sequential walk to binary search over the unconsumed tail. +// +// Per driver ID a walk costs O/D comparisons amortized and a search log2(O), +// so on count alone they cross at O/D = log2(O). Both are cursored, so the +// search skips ground it has covered too, and measured per comparison the walk +// is the more expensive of the two (1.65ns against 1.02ns on scattered +// postings, whose values mispredict the walk's data-dependent branch every +// iteration). That puts the wall-clock crossover at 3 to 6 across driver sizes +// from 4 to 256, and 4 sits on the safe side of a shallow optimum: the walk +// gives up at most 1.28x there, against 4.1x to 4.7x at a ratio of 32. +const mergeRatio = 4 + +// Intersect returns the intersection of ps, or the zero Postings when any +// input is absent or nothing survives. It may reorder ps. +// +// Where the postings come back as ID lists — every term at or below the index's +// delta-coding threshold — intersecting them directly beats materializing them +// into bitmaps first, by 4.3x at cardinality 4, 2.8x at 64 and 1.6x at 1024. +// That is the whole reason the stores hand out un-materialized postings. +// +// The smallest input drives, so the work is bounded by the smallest side and +// the result is at most that large. When every input is already a bitmap there +// is nothing to save and roaring's container-wise FastAnd wins instead. +func Intersect(ps []Postings) Postings { + if len(ps) == 0 { + return Postings{} + } + for _, p := range ps { + if !p.Present() { + return Postings{} + } + } + // See ConcurrentBitmaps.Get: FastAnd Clones a lone input. + if len(ps) == 1 { + return ps[0] + } + + slices.SortFunc(ps, func(a, b Postings) int { + return cmp.Compare(a.Cardinality(), b.Cardinality()) + }) + + if !slices.ContainsFunc(ps, func(p Postings) bool { return p.ids != nil }) { + bms := make([]*roaring.Bitmap, len(ps)) + for i, p := range ps { + bms[i] = p.bm + } + return BitmapPostings(roaring.FastAnd(bms...)) + } + + // The driver is the smallest side, so reading a bitmap one out costs an + // allocation bounded by the smallest cardinality — cheap next to the + // materialization of every other side that this avoids. + driver := ps[0].ids + if driver == nil { + driver = ps[0].bm.ToArray() + } + + probes := make([]prober, len(ps)-1) + for i, p := range ps[1:] { + probes[i] = prober{ + ids: p.ids, + bm: p.bm, + sequential: len(p.ids) <= mergeRatio*len(driver), + } + } + + hits := make([]uint32, 0, len(driver)) +next: + for _, id := range driver { + for i := range probes { + if !probes[i].contains(id) { + continue next + } + } + hits = append(hits, id) + } + return IDPostings(hits) +} + +// prober tests membership in one intersect input. For an ID list it carries a +// cursor, since the driver ascends and so no probe ever needs to look behind +// the last ID it matched. +type prober struct { + ids []uint32 + bm *roaring.Bitmap + cursor int + sequential bool +} + +func (p *prober) contains(id uint32) bool { + if p.ids == nil { + return p.bm.Contains(id) + } + if p.sequential { + for p.cursor < len(p.ids) && p.ids[p.cursor] < id { + p.cursor++ + } + return p.cursor < len(p.ids) && p.ids[p.cursor] == id + } + i, found := slices.BinarySearch(p.ids[p.cursor:], id) + p.cursor += i + return found +} + +// Union returns the union of ps, or the zero Postings when none are present. +// It mutates ps's backing array: present entries compact to the front keeping +// their relative order, and the tail up to the caller's original length is +// zeroed. The caller's own slice header keeps its length — reading ps +// afterwards yields the compacted entries followed by zero-value Postings — +// so treat ps as consumed. +// +// All-list inputs are merged rather than materialized. Building a bitmap from a +// scattered id list costs an allocation per container it touches, which is what +// dominates: 2 filters of 16 postings measure 14.8us and 493 allocations +// through FastOr against 114ns and 2 merged, and 5 filters of 1024 drop from +// 3271 allocations to 9. Keeping the result a list is most of that win, since +// the clip and the selection downstream are far cheaper on one. +// +// Anything else goes to FastOr, which also covers the region where it wins: a +// bitmap input means that term is above the index's delta threshold. +func Union(ps []Postings) Postings { + ps = slices.DeleteFunc(ps, func(p Postings) bool { return !p.Present() }) + switch len(ps) { + case 0: + return Postings{} + case 1: + // See ConcurrentBitmaps.Get: FastOr Clones a lone input. + return ps[0] + } + + if !slices.ContainsFunc(ps, func(p Postings) bool { return p.ids == nil }) { + merged := ps[0].ids + for _, p := range ps[1:] { + merged = mergeAscending(make([]uint32, 0, len(merged)+len(p.ids)), merged, p.ids) + } + return IDPostings(merged) + } + + bms := make([]*roaring.Bitmap, len(ps)) + for i, p := range ps { + bms[i] = p.Bitmap() + } + return BitmapPostings(roaring.FastOr(bms...)) +} + +// mergeAscending appends the union of two ascending deduped lists to dst. +// Merging pairwise beats a k-way heap for the handful of inputs a filter set +// produces: 5 lists of 1024 measure 46us pairwise against 96us through a heap. +func mergeAscending(dst, a, b []uint32) []uint32 { + i, j := 0, 0 + for i < len(a) && j < len(b) { + switch { + case a[i] < b[j]: + dst = append(dst, a[i]) + i++ + case a[i] > b[j]: + dst = append(dst, b[j]) + j++ + default: + dst = append(dst, a[i]) + i++ + j++ + } + } + dst = append(dst, a[i:]...) + return append(dst, b[j:]...) +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/intersect_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/intersect_test.go new file mode 100644 index 000000000..3cfe17538 --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/intersect_test.go @@ -0,0 +1,186 @@ +package event + +import ( + "math/rand" + "slices" + "testing" + + "github.com/RoaringBitmap/roaring/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// makeTestPostings builds card postings scattered over span, in the requested +// form, and returns the equivalent bitmap for cross-checking. +func makeTestPostings(tb testing.TB, rng *rand.Rand, card int, span uint32, asIDs bool) (Postings, *roaring.Bitmap) { + tb.Helper() + // Without this the distinct-value loop below never terminates. + require.LessOrEqual(tb, uint32(card), span, "cardinality must fit in the span") + set := make(map[uint32]struct{}, card) + for len(set) < card { + set[uint32(rng.Int31n(int32(span)))] = struct{}{} + } + ids := make([]uint32, 0, len(set)) + for id := range set { + ids = append(ids, id) + } + slices.Sort(ids) + bm := roaring.BitmapOf(ids...) + if asIDs { + return IDPostings(ids), bm + } + return BitmapPostings(bm), bm +} + +// TestIntersectMatchesFastAnd cross-checks Intersect against roaring.FastAnd, +// the operation it replaces. The span is kept tight relative to cardinality so +// intersections are usually non-empty — with postings spread thinly, most +// rounds would compare one empty result against another and prove nothing. +func TestIntersectMatchesFastAnd(t *testing.T) { + rng := rand.New(rand.NewSource(7)) + + for round := range 300 { + inputs := make([]Postings, 2+rng.Intn(3)) + want := make([]*roaring.Bitmap, len(inputs)) + for i := range inputs { + // Mixed forms and a wide cardinality spread, so the driver is + // sometimes an ID list and sometimes a bitmap, and both probe + // strategies get exercised. + inputs[i], want[i] = makeTestPostings(t, rng, 1+rng.Intn(60), 100, rng.Intn(2) == 0) + } + + got := Intersect(inputs) + expected := roaring.FastAnd(want...) + assert.Equal(t, expected.ToArray(), got.Bitmap().ToArray(), "round %d", round) + assert.Equal(t, !expected.IsEmpty(), got.Present(), "round %d presence", round) + } +} + +// TestIntersectDriverStrategies covers the shapes that select each probe +// strategy, including the ratio at which a cursored walk gives way to binary +// search and the case where the smallest side is a bitmap. +func TestIntersectDriverStrategies(t *testing.T) { + rng := rand.New(rand.NewSource(11)) + + for _, tc := range []struct { + name string + driverCard int + otherCard int + driverAsIDs bool + otherAsIDs bool + }{ + {"equal lists take the walk", 400, 400, true, true}, + {"lopsided lists take the search", 4, 400, true, true}, + {"tiny list driver forces the search", 1, 2000, true, true}, + {"list driver against a bitmap", 40, 2000, true, false}, + {"bitmap driver against a list", 40, 2000, false, true}, + {"all bitmaps fall back to FastAnd", 40, 2000, false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + a, abm := makeTestPostings(t, rng, tc.driverCard, 4000, tc.driverAsIDs) + b, bbm := makeTestPostings(t, rng, tc.otherCard, 4000, tc.otherAsIDs) + + // Pass largest-first: Intersect owns the ordering. + got := Intersect([]Postings{b, a}) + assert.Equal(t, roaring.And(abm, bbm).ToArray(), got.Bitmap().ToArray()) + }) + } +} + +func TestIntersectDegenerate(t *testing.T) { + present := IDPostings([]uint32{1, 2, 3}) + + assert.False(t, Intersect(nil).Present()) + assert.False(t, Intersect([]Postings{}).Present()) + assert.False(t, Intersect([]Postings{present, {}}).Present(), + "one absent term empties the intersection") + assert.False(t, Intersect([]Postings{IDPostings([]uint32{1}), IDPostings([]uint32{2})}).Present(), + "disjoint inputs are absent, not an empty present set") + + // A single input passes through untouched, so a lone bitmap-backed term is + // not materialized on the way past. + bm := roaring.BitmapOf(1, 2, 3) + single := Intersect([]Postings{BitmapPostings(bm)}) + require.True(t, single.Present()) + assert.Same(t, bm, single.Bitmap()) +} + +// TestUnionMatchesFastOr cross-checks the merge path against roaring.FastOr, +// which is both the operation it replaces and the fallback it still uses when +// any input is bitmap-backed. +func TestUnionMatchesFastOr(t *testing.T) { + rng := rand.New(rand.NewSource(13)) + + for round := range 400 { + inputs := make([]Postings, 2+rng.Intn(4)) + want := make([]*roaring.Bitmap, len(inputs)) + // Mixed forms so both the merge and the FastOr fallback run, and + // occasionally all-lists so the merge runs on every input. + allLists := rng.Intn(3) == 0 + for i := range inputs { + asIDs := allLists || rng.Intn(2) == 0 + inputs[i], want[i] = makeTestPostings(t, rng, 1+rng.Intn(80), 200, asIDs) + } + + got := Union(inputs) + expected := roaring.FastOr(want...) + require.True(t, got.Present(), "round %d", round) + assert.Equal(t, expected.ToArray(), got.Bitmap().ToArray(), "round %d", round) + if allLists { + ids := got.IDs() + require.NotNil(t, ids, "round %d: all-list inputs must merge, not materialize", round) + // Assert on the raw slice, not through Bitmap(): that sorts and + // dedups, so it would launder a merge that emitted duplicates or + // went backwards. IDPostings does not check the precondition, and + // violating it answers wrongly rather than failing. + assert.Equal(t, expected.ToArray(), ids, "round %d: merged ids", round) + } + } +} + +// TestUnionMergeBoundaries covers the shapes a randomized sweep reaches rarely: +// full overlap, no overlap, one list a prefix or suffix of the other, and +// duplicate values across every input. +func TestUnionMergeBoundaries(t *testing.T) { + for _, tc := range []struct { + name string + in [][]uint32 + want []uint32 + }{ + {"identical", [][]uint32{{1, 2, 3}, {1, 2, 3}}, []uint32{1, 2, 3}}, + {"disjoint interleaved", [][]uint32{{1, 3, 5}, {2, 4, 6}}, []uint32{1, 2, 3, 4, 5, 6}}, + {"disjoint blocks", [][]uint32{{1, 2}, {8, 9}}, []uint32{1, 2, 8, 9}}, + {"prefix", [][]uint32{{1, 2}, {1, 2, 3, 4}}, []uint32{1, 2, 3, 4}}, + {"suffix", [][]uint32{{3, 4}, {1, 2, 3, 4}}, []uint32{1, 2, 3, 4}}, + {"singletons", [][]uint32{{5}, {5}, {5}}, []uint32{5}}, + {"three-way", [][]uint32{{1, 4}, {2, 5}, {3, 6}}, []uint32{1, 2, 3, 4, 5, 6}}, + {"extremes", [][]uint32{{0}, {4294967295}}, []uint32{0, 4294967295}}, + } { + t.Run(tc.name, func(t *testing.T) { + ps := make([]Postings, len(tc.in)) + for i, ids := range tc.in { + ps[i] = IDPostings(ids) + } + got := Union(ps) + require.NotNil(t, got.IDs(), "all-list inputs must merge") + assert.Equal(t, tc.want, got.SelectIDs(0, false)) + }) + } +} + +func TestUnion(t *testing.T) { + a := IDPostings([]uint32{1, 5}) + b := BitmapPostings(roaring.BitmapOf(5, 9)) + + assert.False(t, Union(nil).Present()) + assert.False(t, Union([]Postings{{}, {}}).Present()) + + // A lone present input passes through, keeping its form. + single := Union([]Postings{{}, a, {}}) + require.True(t, single.Present()) + assert.NotNil(t, single.IDs(), "a passed-through input must keep its form") + + both := Union([]Postings{a, b}) + require.True(t, both.Present()) + assert.Equal(t, []uint32{1, 5, 9}, both.SelectIDs(0, false)) +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/match.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/match.go index 77563b14b..9a13c383a 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/match.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/match.go @@ -13,14 +13,11 @@ package event import ( "bytes" - "cmp" "context" "fmt" "iter" "slices" - "github.com/RoaringBitmap/roaring/v2" - protocol "github.com/stellar/go-stellar-sdk/protocols/rpc" "github.com/stellar/go-stellar-sdk/xdr" ) @@ -107,7 +104,7 @@ func (f *Filter) valueTermKeys() []TermKey { } // termGroups returns the indexed terms this filter constrains, grouped -// by field: the bitmaps within a group are OR-ed and the groups are +// by field: the postings within a group are OR-ed and the groups are // AND-ed. Only the topic-count group ever holds more than one term. func (f *Filter) termGroups() [][]TermKey { var groups [][]TermKey @@ -280,7 +277,7 @@ func Matches( streamRange(ctx, r, window, descending, firstBatch, yield) return } - if union.IsEmpty() { + if !union.Present() { return } streamUnion(ctx, r, filters, union, descending, firstBatch, yield) @@ -316,16 +313,19 @@ func validateMatchCall(ctx context.Context, r Reader, filters []Filter, window I // numbered steps below). matchAll reports that some filter (or the // empty slice) constrains nothing, detected before any index I/O; the // caller then streams the window directly. Otherwise the result is -// empty when no candidate falls in the window, and is never a -// borrowed mirror snapshot (the window AND allocates on the borrowing -// path), so downstream iteration is safe. +// absent when no candidate falls in the window. +// +// The result may be BORROWED store state: Intersect and Union both pass a +// lone input through, and ClipRange can return its receiver. That is safe +// because nothing downstream writes to it (see Postings), and +// FetchEvents takes the ids read-only. func unionForFilters( ctx context.Context, r Reader, filters []Filter, window IDRange, -) (*roaring.Bitmap, bool, error) { +) (Postings, bool, error) { // ───── 1. Dedupe terms across filters ───── // // filterPlans[i] holds the slots filter i needs out of the batched - // lookup: the bitmaps within a group are OR-ed, the groups AND-ed. + // lookup: the postings within a group are OR-ed, the groups AND-ed. // // A filter that asks the index for no terms constrains nothing, and // so does an empty filter slice: both take the match-all path. @@ -333,7 +333,7 @@ func unionForFilters( // an unconstrained filter from intersecting nothing and coming back // empty instead. if len(filters) == 0 { - return nil, true, nil + return Postings{}, true, nil } filterPlans := make([]termPlan, len(filters)) var uniqueKeys []TermKey @@ -341,7 +341,7 @@ func unionForFilters( for i := range filters { groups := filters[i].termGroups() if len(groups) == 0 { - return nil, true, nil + return Postings{}, true, nil } plan := make(termPlan, len(groups)) for g, keys := range groups { @@ -355,33 +355,28 @@ func unionForFilters( } // ───── 2. Single batched lookup for all unique terms ───── - bitmaps, err := r.LookupKeys(ctx, uniqueKeys) + postings, err := r.LookupKeys(ctx, uniqueKeys) if err != nil { - return nil, false, fmt.Errorf("events: query lookup: %w", err) + return Postings{}, false, fmt.Errorf("events: query lookup: %w", err) } // ───── 3. Per-filter intersect ───── // - // If a whole group is absent from the index (every bitmap in it is - // nil), that filter's intersection is empty — skip it without - // contributing to the union. + // If a whole group is absent from the index, that filter's + // intersection is empty — skip it without contributing to the union, + // and without unioning the groups after it. // - // Bitmap ownership in perFilter is mixed: - // - Single-constraint filter: we borrow bitmaps[s] directly (a - // mirror snapshot from LookupKeys), skipping FastAnd's Clone. - // - Multi-constraint filter: FastAnd allocates a fresh result. - // Either way the downstream union (FastOr) and the window AND never - // mutate their inputs, so a borrowed entry stays valid through - // the rest of the function. FastAnd never mutates its inputs - // either, so the same bitmap may appear across multiple filters - // safely. - perFilter := make([]*roaring.Bitmap, 0, len(filterPlans)) + // Intersect drives from the smallest side whatever form it is in and + // probes the rest, rather than materializing every side into a bitmap + // first. Entries here may be borrowed store state and stay borrowed to + // the end; see unionForFilters' contract. + perFilter := make([]Postings, 0, len(filterPlans)) for _, plan := range filterPlans { - inputs := make([]*roaring.Bitmap, 0, len(plan)) + inputs := make([]Postings, 0, len(plan)) missed := false for _, slots := range plan { - group := unionSlots(bitmaps, slots) - if group == nil { + group := unionSlots(postings, slots) + if !group.Present() { missed = true break } @@ -390,59 +385,39 @@ func unionForFilters( if missed { continue } - if len(inputs) == 1 { - perFilter = append(perFilter, inputs[0]) - continue + // Intersect sorts what it is given, so inputs must be this + // filter's own slice — postings itself is indexed by slot and + // shared across filters. + if hits := Intersect(inputs); hits.Present() { + perFilter = append(perFilter, hits) } - // FastAnd intersects left-to-right — putting the smallest - // bitmap first shrinks the accumulator fastest. roaring's own - // docs call this out as the recommended caller-side prep. - slices.SortFunc(inputs, func(a, b *roaring.Bitmap) int { - return cmp.Compare(a.GetCardinality(), b.GetCardinality()) - }) - perFilter = append(perFilter, roaring.FastAnd(inputs...)) } if len(perFilter) == 0 { - return roaring.New(), false, nil + return Postings{}, false, nil } // ───── 4. Union across filters ───── - // Single-filter case: FastOr would Clone — skip it and use the - // already-computed bitmap directly. That bitmap may be borrowed - // (from LookupKeys), so step 5's window And uses the fresh-result - // variant on that path to avoid mutating shared state. - var union *roaring.Bitmap - singleFilter := len(perFilter) == 1 - if singleFilter { - union = perFilter[0] - } else { - union = roaring.FastOr(perFilter...) - } + // Union merges ascending lists rather than building a bitmap per + // input, and passes a lone input through untouched. + union := Union(perFilter) // ───── 5. Apply the event-ID window ───── // - // The window AND enforces the caller's pinned range. It also clips + // The clip enforces the caller's pinned range. It also drops // phantom IDs from a concurrent hot-store ingest: the mirror // publishes entries before offsets, so LookupKeys can briefly - // surface IDs past EventCount. The AND keeps the stream strictly + // surface IDs past EventCount. It keeps the stream strictly // within the snapshot the caller pinned at request entry. // - // This covers the multi-term group too. Its bitmaps are separate + // This covers the multi-term group too. Its postings are separate // mirror snapshots taken at different instants, but an event never // moves between the terms of one group once ingested, so a torn read // across them can only surface IDs past the pinned End. - rangeBM := roaring.New() - rangeBM.AddRange(uint64(window.Start), uint64(window.End)) - if singleFilter { - union = roaring.And(union, rangeBM) // fresh result; union may be borrowed - } else { - union.And(rangeBM) // FastOr output is owned; in-place is fine - } - return union, false, nil + return union.ClipRange(window.Start, window.End), false, nil } -// streamUnion walks the union bitmap in internal batches: collect +// streamUnion walks the union in internal batches: collect // candidate ordinals up to the batch size, fetch, post-filter, yield // the survivors. Drops advance the walk with no yield. The first // batch is sized to firstBatch (see Matches); later batches use the @@ -453,17 +428,23 @@ func unionForFilters( // fetched matches are flipped back. Stepping one id at a time is fine // here: the fetch I/O dominates a 512-step loop. func streamUnion( - ctx context.Context, r Reader, filters []Filter, union *roaring.Bitmap, + ctx context.Context, r Reader, filters []Filter, union Postings, descending bool, firstBatch int, yield func(Match, error) bool, ) { var it interface { HasNext() bool Next() uint32 } - if descending { - it = union.ReverseIterator() - } else { - it = union.Iterator() + switch ids := union.IDs(); { + case ids != nil: + // An ID-backed union — every term the cold index stored below its + // delta threshold, and every sparse hot term — is walked over its + // own slice, so the stream never materializes a bitmap for it. + it = newIDCursor(ids, descending) + case descending: + it = union.Bitmap().ReverseIterator() + default: + it = union.Bitmap().Iterator() } batch, rest := batchSizes(firstBatch) ids := make([]uint32, 0, batch) @@ -562,28 +543,44 @@ func CountDistinctTerms(filters []Filter) int { return len(unique) } -// unionSlots ORs the bitmaps at slots, and returns nil when every one -// of them is absent from the index. A lone present bitmap is borrowed -// rather than cloned, like the single-constraint path in -// unionForFilters. -func unionSlots(bitmaps []*roaring.Bitmap, slots []int) *roaring.Bitmap { +// unionSlots ORs the postings at slots, and returns the zero Postings +// when every one of them is absent from the index. A lone present entry +// is borrowed rather than copied — Union passes a single input through. +func unionSlots(postings []Postings, slots []int) Postings { if len(slots) == 1 { - return bitmaps[slots[0]] + return postings[slots[0]] } - present := make([]*roaring.Bitmap, 0, len(slots)) + // Union compacts the slice it is given in place, so this must be a + // fresh one rather than a view of the slot-indexed result. + group := make([]Postings, 0, len(slots)) for _, s := range slots { - if bitmaps[s] != nil { - present = append(present, bitmaps[s]) - } + group = append(group, postings[s]) } - switch len(present) { - case 0: - return nil - case 1: - return present[0] - default: - return roaring.FastOr(present...) + return Union(group) +} + +// idCursor walks an ascending id slice in stream order, matching the +// HasNext/Next shape of roaring's iterators so streamUnion does not have to +// branch per id. +type idCursor struct { + ids []uint32 + next int + step int +} + +func newIDCursor(ids []uint32, descending bool) *idCursor { + if descending { + return &idCursor{ids: ids, next: len(ids) - 1, step: -1} } + return &idCursor{ids: ids, step: 1} +} + +func (c *idCursor) HasNext() bool { return c.next >= 0 && c.next < len(c.ids) } + +func (c *idCursor) Next() uint32 { + id := c.ids[c.next] + c.next += c.step + return id } // indexOfOrAddTerm returns the index of key inside *keys, appending 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 671bc2718..2be87c5c2 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/match_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/match_test.go @@ -54,7 +54,7 @@ type countingReader struct { totalKeys int } -func (c *countingReader) LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring.Bitmap, error) { +func (c *countingReader) LookupKeys(ctx context.Context, keys []TermKey) ([]Postings, error) { c.lookupKeysCalls++ c.totalKeys += len(keys) return c.Reader.LookupKeys(ctx, keys) @@ -1131,19 +1131,22 @@ func TestQuery_InvalidFilterRejected(t *testing.T) { // multi-term group, and the overflow bucket is populated in any chunk holding // an event with topics. func TestUnionSlots(t *testing.T) { - first := roaring.BitmapOf(1, 2) - second := roaring.BitmapOf(3) - bitmaps := []*roaring.Bitmap{first, nil, second, nil} - - assert.Same(t, first, unionSlots(bitmaps, []int{0}), - "a lone bitmap is borrowed, not cloned") - assert.Nil(t, unionSlots(bitmaps, []int{1})) - assert.Nil(t, unionSlots(bitmaps, []int{1, 3}), + firstIDs := []uint32{1, 2} + first := IDPostings(firstIDs) + second := BitmapPostings(roaring.BitmapOf(3)) + postings := []Postings{first, {}, second, {}} + + assert.Equal(t, firstIDs, unionSlots(postings, []int{0}).IDs(), + "a lone entry is borrowed, not copied") + assert.False(t, unionSlots(postings, []int{1}).Present()) + assert.False(t, unionSlots(postings, []int{1, 3}).Present(), "a group absent from the index empties the filter") - assert.Same(t, second, unionSlots(bitmaps, []int{1, 2}), - "the one present bitmap in a group is borrowed too") - assert.Equal(t, []uint32{1, 2, 3}, unionSlots(bitmaps, []int{0, 2}).ToArray()) - assert.Equal(t, []uint32{1, 2}, first.ToArray(), "inputs must not be mutated") + assert.Same(t, second.Bitmap(), unionSlots(postings, []int{1, 2}).Bitmap(), + "the one present entry in a group is borrowed too") + assert.Equal(t, []uint32{1, 2, 3}, unionSlots(postings, []int{0, 2}).Bitmap().ToArray()) + assert.Equal(t, []uint32{1, 2}, firstIDs, "inputs must not be mutated") + assert.Equal(t, []uint32{1, 2}, postings[0].IDs(), + "Union must not compact the caller's slot-indexed results") } // ─── Cold-reader parity coverage ──────────────────────────────────────── diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/postings.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings.go new file mode 100644 index 000000000..aeb9a36be --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings.go @@ -0,0 +1,163 @@ +package event + +import ( + "slices" + + "github.com/RoaringBitmap/roaring/v2" +) + +// Postings is one term's event IDs in whichever form the store already had +// them: an ascending ID slice for a delta-coded term, a bitmap for a roaring +// one. Intersect explains why the un-materialized form is worth carrying. +// +// The zero value means the term is absent. A present term always holds at +// least one posting, so that cannot be confused with an empty set. +// +// Treat a Postings and anything it returns as READ-ONLY: a bitmap-backed one +// may be a store's live mirror snapshot, so mutating it corrupts the store. +type Postings struct { + ids []uint32 + bm *roaring.Bitmap +} + +// IDPostings wraps an ID slice, which MUST be strictly ascending. That is not +// checked, and violating it does not error: ClipRange binary-searches the slice +// and Intersect walks it with a cursor that never looks back, so the result is +// silently wrong. +// +// An empty slice is absent. +func IDPostings(ids []uint32) Postings { + if len(ids) == 0 { + return Postings{} + } + return Postings{ids: ids} +} + +// BitmapPostings wraps a bitmap. A nil or empty bitmap is absent. +func BitmapPostings(bm *roaring.Bitmap) Postings { + if bm == nil || bm.IsEmpty() { + return Postings{} + } + return Postings{bm: bm} +} + +// Present reports whether the term was found. +func (p Postings) Present() bool { return p.ids != nil || p.bm != nil } + +// Cardinality returns the number of postings. +func (p Postings) Cardinality() uint64 { + if p.ids != nil { + return uint64(len(p.ids)) + } + if p.bm != nil { + return p.bm.GetCardinality() + } + return 0 +} + +// IDs returns the backing ID slice, or nil when this Postings is bitmap-backed. +func (p Postings) IDs() []uint32 { return p.ids } + +// Contains reports membership. +func (p Postings) Contains(id uint32) bool { + if p.ids != nil { + _, ok := slices.BinarySearch(p.ids, id) + return ok + } + return p.bm != nil && p.bm.Contains(id) +} + +// Bitmap returns a read-only bitmap over these postings. A bitmap-backed +// Postings hands back the store's own; an ID-backed one materializes. +func (p Postings) Bitmap() *roaring.Bitmap { + if p.bm != nil { + return p.bm + } + return roaring.BitmapOf(p.ids...) +} + +// ClipRange returns the postings that fall in [start, end). The result is +// read-only like the receiver, and may share its backing memory. +func (p Postings) ClipRange(start, end uint32) Postings { + if start >= end || !p.Present() { + return Postings{} + } + if p.ids != nil { + lo, _ := slices.BinarySearch(p.ids, start) + hi, _ := slices.BinarySearch(p.ids, end) + // Capacity is clamped so a downstream append cannot overwrite a + // posting another reader can still see. + return IDPostings(p.ids[lo:hi:hi]) + } + if p.bm.GetCardinality() == 0 { + // Minimum and Maximum fault on a bitmap that holds containers but no + // postings. Producers do not make one and the cold decode rejects + // them, so this is insurance, paid once per query rather than per + // term. + return Postings{} + } + if p.bm.Minimum() >= start && p.bm.Maximum() < end { + // Nothing to clip. Worth checking because the window bitmap below + // costs O(range width) rather than O(cardinality) — spanning a whole + // chunk it is ~10us and 286 allocations, whatever the postings are — + // and a full-range query is the common one. + return p + } + window := roaring.New() + window.AddRange(uint64(start), uint64(end)) + // Fresh result: p.bm may be the store's live snapshot, so it cannot be + // intersected in place. + return BitmapPostings(roaring.And(p.bm, window)) +} + +// SelectIDs returns up to maxIDs postings in ascending order, keeping the +// lowest when descending is false and the highest when it is true. maxIDs <= 0 +// returns all of them. The result is read-only and may share the receiver's +// memory. +// +// The cap lands here rather than after the caller's own filtering, so a short +// result does not mean the postings are exhausted. +func (p Postings) SelectIDs(maxIDs int, descending bool) []uint32 { + if p.ids != nil { + // Capacities are clamped for the reason ClipRange gives. + n := len(p.ids) + if maxIDs <= 0 || maxIDs >= n { + return p.ids[:n:n] + } + if descending { + return p.ids[n-maxIDs : n : n] + } + return p.ids[:maxIDs:maxIDs] + } + if p.bm == nil { + return nil + } + + card := p.bm.GetCardinality() + limit := card + if maxIDs > 0 && uint64(maxIDs) < limit { + limit = uint64(maxIDs) + } + // limit <= card (a chunk's event-id space, max ~10M today) and <= maxIDs + // (int), so the int conversions below are safe. + if descending && limit < card { + ids := make([]uint32, 0, limit) + ri := p.bm.ReverseIterator() + for ri.HasNext() && uint64(len(ids)) < limit { + ids = append(ids, ri.Next()) + } + slices.Reverse(ids) // reverse-iterator order is descending, we want ascending + return ids + } + ids := make([]uint32, limit) + mi := p.bm.ManyIterator() + filled := 0 + for filled < int(limit) { //nolint:gosec // limit <= maxIDs (int) <= MaxInt + n := mi.NextMany(ids[filled:]) + if n == 0 { + break + } + filled += n + } + return ids[:filled] +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_codec.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_codec.go new file mode 100644 index 000000000..3e855105d --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_codec.go @@ -0,0 +1,129 @@ +package event + +import ( + "encoding/binary" + "errors" + "fmt" + "math" +) + +// This file is the postings codec: the byte encoding for ONE term's ascending +// event IDs, without any term prefix. +// +// uvarint id-count ‖ ids as uvarints +// (first absolute, then strictly-positive deltas) +// +// Event IDs are strictly ascending per term, so deltas are always ≥ 1 — the +// decoder rejects a zero delta as corruption. index.pack's delta codec +// (itemCodecDelta, see stores/event/cold_format.go) is exactly this encoding +// behind a codec byte. + +// AppendPostings appends one term's ID list — uvarint count, then the IDs +// (first absolute, then strictly-positive deltas) — WITHOUT any term prefix, +// for consumers that already know which term they are reading. ids must be +// strictly ascending. +func AppendPostings(dst []byte, ids []uint32) []byte { + dst = binary.AppendUvarint(dst, uint64(len(ids))) + var prev uint32 + for j, id := range ids { + if j == 0 { + dst = binary.AppendUvarint(dst, uint64(id)) + } else { + dst = binary.AppendUvarint(dst, uint64(id-prev)) + } + prev = id + } + return dst +} + +// decodePostings decodes the posting list at the head of b — uvarint count, +// then the IDs — into ids, and returns the unconsumed remainder. Errors are +// unwrapped; callers add their own context, matching DecodeAscendingIDs. +func decodePostings(b []byte, ids []uint32) ([]uint32, []byte, error) { + count, n := binary.Uvarint(b) + if n <= 0 { + return nil, nil, errors.New("bad id-count uvarint") + } + b = b[n:] + // Each ID takes ≥1 byte, so a count beyond the remaining bytes is + // structurally impossible — reject before allocating for it. + if count == 0 || count > uint64(len(b)) { + return nil, nil, fmt.Errorf("id count %d exceeds %d remaining bytes", count, len(b)) + } + next := func() (uint64, error) { + v, k := binary.Uvarint(b) + if k <= 0 { + return 0, errors.New("bad event-id uvarint") + } + b = b[k:] + return v, nil + } + // Grow once rather than letting append double its way there. count is + // bounded by the remaining bytes above, so it cannot force a large + // allocation from a small record. + if uint64(cap(ids)) < count { + ids = make([]uint32, 0, count) + } + out, err := DecodeAscendingIDs(next, count, ids[:0]) + if err != nil { + return nil, nil, err + } + return out, b, nil +} + +// DecodeAscendingIDs decodes count event IDs (first absolute, then +// strictly-positive deltas) into ids, drawing raw uvarints from next. It is +// THE validation core for the ID stream — every decoder delegates here, so the +// security-relevant invariants (raw-varint reject BEFORE accumulation, +// zero-delta reject, uint32 overflow) cannot drift between decoders. Errors +// are unwrapped; callers add their own context. +// +// It takes a next function rather than a slice because a streaming decoder +// over an io.Reader has to share these checks with the slice-based +// decodePostings above. +func DecodeAscendingIDs(next func() (uint64, error), count uint64, ids []uint32) ([]uint32, error) { + var prev uint64 + for i := range count { + v, err := next() + if err != nil { + return nil, err + } + // Reject the raw varint before accumulating: a crafted delta near + // 2^64 would wrap prev+v back under MaxUint32 and smuggle a + // non-ascending ID past the overflow check (postings are untrusted). + if v > math.MaxUint32 { + return nil, fmt.Errorf("delta %d overflows uint32", v) + } + abs := v + if i > 0 { + if v == 0 { + return nil, errors.New("zero delta") + } + abs = prev + v + } + if abs > math.MaxUint32 { + return nil, fmt.Errorf("event id %d overflows uint32", abs) + } + ids = append(ids, uint32(abs)) + prev = abs + } + return ids, nil +} + +// DecodePostings decodes one term's ID list as written by AppendPostings. b +// must hold exactly one posting list and nothing else: trailing bytes are +// rejected, because the cold index sizes its records from the writer and any +// slack means the record was not written by this codec. +// +// The result is freshly allocated, so it can outlive the buffer it was decoded +// from and cannot alias a sibling decode. +func DecodePostings(b []byte) ([]uint32, error) { + out, rest, err := decodePostings(b, nil) + if err != nil { + return nil, fmt.Errorf("events: postings: %w", err) + } + if len(rest) != 0 { + return nil, fmt.Errorf("events: postings: %d trailing bytes", len(rest)) + } + return out, nil +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_codec_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_codec_test.go new file mode 100644 index 000000000..26af35685 --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_codec_test.go @@ -0,0 +1,74 @@ +package event + +import ( + "encoding/binary" + "math" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPostingsCodec_RoundTrip(t *testing.T) { + for _, tc := range []struct { + name string + ids []uint32 + }{ + {"one posting", []uint32{0}}, + {"contiguous", []uint32{0, 1, 2, 3}}, + // Non-contiguous, large absolute IDs exercise multi-byte varints. + {"scattered", []uint32{7, 1 << 20, 1<<20 + 1}}, + {"extremes", []uint32{0, math.MaxUint32}}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := DecodePostings(AppendPostings(nil, tc.ids)) + require.NoError(t, err) + assert.Equal(t, tc.ids, got) + }) + } +} + +// TestPostingsCodec_Appends pins that AppendPostings appends rather than +// replaces: index.pack's writer puts the codec byte down first and encodes +// straight after it. +func TestPostingsCodec_Appends(t *testing.T) { + buf := AppendPostings([]byte{0xff}, []uint32{3, 4}) + require.Equal(t, byte(0xff), buf[0]) + got, err := DecodePostings(buf[1:]) + require.NoError(t, err) + assert.Equal(t, []uint32{3, 4}, got) +} + +func TestPostingsCodec_RejectsCorruption(t *testing.T) { + good := AppendPostings(nil, []uint32{3, 4}) + + // Empty body: no count uvarint at all. + require.ErrorContains(t, decodeErr(t, nil), "bad id-count uvarint") + // A zero count cannot describe a term: a stored term always holds at + // least one posting. + require.ErrorContains(t, decodeErr(t, []byte{0}), "exceeds") + // id count exceeding the remaining bytes (truncate the ID area). + require.ErrorContains(t, decodeErr(t, good[:len(good)-1]), "exceeds") + // Zero delta = duplicate ID within a term. + require.ErrorContains(t, decodeErr(t, []byte{2 /* count */, 3 /* first id */, 0}), "zero delta") + // Trailing bytes mean the record was not written by this codec. + require.ErrorContains(t, decodeErr(t, append(good, 0x00)), "trailing bytes") + + // A delta that overflows uint32 must be rejected as a raw varint, before + // it is added to the running absolute: a value near 2^64 would otherwise + // wrap back under MaxUint32 and smuggle in a non-ascending ID. + wrap := binary.AppendUvarint([]byte{2 /* count */, 1 /* first id */}, math.MaxUint64) + require.ErrorContains(t, decodeErr(t, wrap), "overflows uint32") + + // An in-range delta whose running absolute leaves uint32 is rejected too. + over := binary.AppendUvarint([]byte{2 /* count */}, math.MaxUint32) + over = binary.AppendUvarint(over, 1) + require.ErrorContains(t, decodeErr(t, over), "overflows uint32") +} + +func decodeErr(t *testing.T, b []byte) error { + t.Helper() + _, err := DecodePostings(b) + require.Error(t, err) + return err +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_test.go new file mode 100644 index 000000000..3dac609bd --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/postings_test.go @@ -0,0 +1,169 @@ +package event + +import ( + "encoding/hex" + "testing" + + "github.com/RoaringBitmap/roaring/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPostingsForms(t *testing.T) { + ids := []uint32{3, 9, 40, 41, 9000} + + for _, tc := range []struct { + name string + post Postings + wantIDs bool + }{ + {"ids", IDPostings(ids), true}, + {"bitmap", BitmapPostings(roaring.BitmapOf(ids...)), false}, + } { + t.Run(tc.name, func(t *testing.T) { + require.True(t, tc.post.Present()) + assert.Equal(t, uint64(len(ids)), tc.post.Cardinality()) + assert.Equal(t, tc.wantIDs, tc.post.IDs() != nil, + "IDs() is the planner's signal that it can drive from this side") + + for _, id := range ids { + assert.True(t, tc.post.Contains(id), "missing %d", id) + } + for _, id := range []uint32{0, 2, 4, 39, 42, 8999, 9001} { + assert.False(t, tc.post.Contains(id), "unexpected %d", id) + } + + assert.Equal(t, ids, tc.post.Bitmap().ToArray()) + }) + } +} + +// TestPostingsAbsent pins the zero value as the miss signal, which is what +// LookupKeys returns for a key the index does not hold. +func TestPostingsAbsent(t *testing.T) { + for _, tc := range []struct { + name string + post Postings + }{ + {"zero value", Postings{}}, + {"nil ids", IDPostings(nil)}, + {"empty ids", IDPostings([]uint32{})}, + {"nil bitmap", BitmapPostings(nil)}, + {"empty bitmap", BitmapPostings(roaring.New())}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.False(t, tc.post.Present()) + assert.Zero(t, tc.post.Cardinality()) + assert.False(t, tc.post.Contains(0)) + assert.True(t, tc.post.Bitmap().IsEmpty()) + }) + } +} + +// TestPostingsBitmapBorrows pins that a bitmap-backed Postings hands back the +// store's own bitmap rather than a copy. The hot store's dense overlay relies +// on that to answer without a per-key Clone, which is also why callers must +// treat the result as read-only. +func TestPostingsBitmapBorrows(t *testing.T) { + bm := roaring.BitmapOf(1, 2, 3) + assert.Same(t, bm, BitmapPostings(bm).Bitmap()) + + // An ID-backed one has nothing to borrow, so it materializes fresh and + // mutating the result cannot reach the postings. + ids := []uint32{1, 2, 3} + p := IDPostings(ids) + p.Bitmap().Add(4) + assert.False(t, p.Contains(4)) + assert.Equal(t, []uint32{1, 2, 3}, ids) +} + +// TestPostingsClipRangeAndSelect checks both forms agree, since the ID form +// takes a completely different route: two binary searches and a subslice +// against a window bitmap and an intersection. +func TestPostingsClipRangeAndSelect(t *testing.T) { + ids := []uint32{3, 9, 40, 41, 9000} + + forms := map[string]Postings{ + "ids": IDPostings(ids), + "bitmap": BitmapPostings(roaring.BitmapOf(ids...)), + } + + for _, tc := range []struct { + name string + start, end uint32 + want []uint32 + }{ + {"everything", 0, 10000, ids}, + {"exact bounds are half-open", 3, 9000, []uint32{3, 9, 40, 41}}, + {"interior", 9, 42, []uint32{9, 40, 41}}, + {"between postings", 42, 8999, nil}, + {"empty range", 40, 40, nil}, + {"inverted range", 100, 10, nil}, + {"below everything", 0, 3, nil}, + {"above everything", 9001, 99999, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + for form, p := range forms { + got := p.ClipRange(tc.start, tc.end) + if tc.want == nil { + assert.False(t, got.Present(), form) + continue + } + require.True(t, got.Present(), form) + assert.Equal(t, tc.want, got.SelectIDs(0, false), form) + } + }) + } + + for _, tc := range []struct { + name string + maxIDs int + descending bool + want []uint32 + }{ + {"uncapped", 0, false, ids}, + {"uncapped descending is still ascending", 0, true, ids}, + {"cap keeps the lowest", 2, false, []uint32{3, 9}}, + {"descending cap keeps the highest", 2, true, []uint32{41, 9000}}, + {"cap above cardinality", 99, false, ids}, + {"negative cap means all", -1, false, ids}, + } { + t.Run(tc.name, func(t *testing.T) { + for form, p := range forms { + assert.Equal(t, tc.want, p.SelectIDs(tc.maxIDs, tc.descending), form) + } + }) + } +} + +// emptyRunContainerBitmap is a bitmap roaring's UnmarshalBinary accepts but +// that holds containers and no postings: one run container with zero +// intervals. Minimum and Maximum fault on it. +func emptyRunContainerBitmap(t *testing.T) *roaring.Bitmap { + t.Helper() + raw, err := hex.DecodeString("3b3000000100008713000000008713") + require.NoError(t, err) + bm := roaring.New() + require.NoError(t, bm.UnmarshalBinary(raw), "roaring must accept it, else this test proves nothing") + require.False(t, bm.IsEmpty(), "must have containers") + require.Zero(t, bm.GetCardinality(), "and no postings") + return bm +} + +// TestPostingsClipRangeToleratesPostingless pins that ClipRange does not fault +// on a bitmap with containers but no postings. Producers never make one, and +// the cold reader rejects it at decode, so this is the last line rather than +// the first. +func TestPostingsClipRangeToleratesPostingless(t *testing.T) { + p := BitmapPostings(emptyRunContainerBitmap(t)) + + assert.NotPanics(t, func() { + assert.False(t, p.ClipRange(0, 1_000_000).Present()) + }) + assert.NotPanics(t, func() { + assert.Empty(t, p.SelectIDs(0, false)) + }) + assert.NotPanics(t, func() { + assert.False(t, Intersect([]Postings{p, IDPostings([]uint32{1})}).Present()) + }) +} diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go index 9c72fb865..4ffe87627 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/reader.go @@ -11,8 +11,6 @@ import ( "fmt" "iter" - "github.com/RoaringBitmap/roaring/v2" - "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/chunk" ) @@ -90,46 +88,50 @@ type Reader interface { // Returns (nil, stores.ErrStoreClosed) after Close. Offsets() (*LedgerOffsets, error) - // LookupKeys returns bitmaps for each key, aligned positionally + // LookupKeys returns each key's postings, 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 the zero Postings if keys[i] has no + // matching events in this chunk — a per-key miss is not an error. + // + // A term the cold index stored below its delta threshold comes + // back as an ascending id slice, not a bitmap; see Postings + // and Intersect for why the planner keeps it that way. + // + // Ownership: callers MUST treat the result as read-only. Some of + // what comes back is the hot store's live state, either a bitmap + // snapshot or the store's own id slice, so writing to it corrupts + // the store for every other reader. The cold path returns freshly + // decoded postings logically owned by the caller, but nothing + // downstream is allowed to tell the difference. // // ColdReader coalesces the underlying packfile reads into a // single ReadItems pass, fanning out across the worker count - // configured via ColdReaderOptions.Concurrency. HotStore returns - // borrowed mirror references with no per-key Clone. - // - // Bitmap ownership: callers MUST treat returned bitmaps as - // read-only. The hot path returns immutable snapshots of the - // live mirror — ConcurrentBitmaps stores bitmap pointers via - // atomic.Pointer COW, so a returned pointer will never be - // mutated by anyone. The cold path returns freshly-unmarshaled - // bitmaps logically owned by the caller. Either way callers - // must not mutate; event.Matches is the only consumer today - // and never mutates, and downstream roaring.FastAnd/FastOr never - // mutate inputs. + // configured via ColdReaderOptions.Concurrency. // // 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. - LookupKeys(ctx context.Context, keys []TermKey) ([]*roaring.Bitmap, error) + LookupKeys(ctx context.Context, keys []TermKey) ([]Postings, error) // FetchEvents decodes events for the supplied chunk-relative // eventIDs and returns them positionally aligned with the input // slice (result[i] corresponds to eventIDs[i]). // + // eventIDs is READ-ONLY: a caller may hand over a window onto the + // hot store's published postings, so sorting or appending in place + // would corrupt live store state, not just this call. + // // eventIDs MUST be sorted ascending with no duplicates. Matches - // iterating a bitmap intersection (roaring.Bitmap.Iterator yields - // ascending) satisfies this for free. Both implementations - // validate the precondition up front and return a wrapped - // ErrUnsortedEventIDs on violation. + // collects them from an ascending walk of the union, so it + // satisfies this for free. Both implementations validate the + // precondition up front and return a wrapped ErrUnsortedEventIDs + // on violation. // // ctx cancels in-flight I/O; the cold path checks ctx between // scattered-read batches, the hot path checks between Gets. // // A missing row is an error: every caller passes ids that name - // stored events (bitmap ids come from LookupKeys; the pager's + // stored events (candidate ids come from LookupKeys; the pager's // resume ordinal is bounds-checked against its ledger's ID range // first), so a miss signals corruption or a writer/reader // mismatch, not a normal not-found case. diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk_test.go index 67e1494bf..f5d6bcacd 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk_test.go @@ -157,10 +157,10 @@ func TestIngestLedger_AllCFsAdvanceTogether(t *testing.T) { require.NoError(t, err) assert.Equal(t, first+1, seqB) // events CFs. - bms, err := db.Events().LookupKeys(context.Background(), []event.TermKey{termA}) + post, err := db.Events().LookupKeys(context.Background(), []event.TermKey{termA}) require.NoError(t, err) - require.NotNil(t, bms[0]) - assert.Equal(t, uint64(2), bms[0].GetCardinality(), "both ledgers share the event term") + require.True(t, post[0].Present()) + assert.Equal(t, uint64(2), post[0].Cardinality(), "both ledgers share the event term") assert.Equal(t, uint32(2), eventCount(t, db.Events())) // The single authoritative committed frontier equals the last committed seq. @@ -196,10 +196,10 @@ func TestIngestLedger_RejectedLedgerPersistsNothingAcrossAnyCF(t *testing.T) { // txhash CFs — the hash is absent. _, gerr = db.Txhash().Get(hash) require.ErrorIs(t, gerr, stores.ErrNotFound) - // events CFs — no term indexed, no event committed (clean miss = nil bitmap). + // events CFs — no term indexed, no event committed (clean miss = absent). bms, lerr := db.Events().LookupKeys(context.Background(), []event.TermKey{term}) require.NoError(t, lerr) - require.Nil(t, bms[0]) + require.False(t, bms[0].Present()) assert.Equal(t, uint32(0), eventCount(t, db.Events())) // The single committed frontier is still empty — nothing committed. @@ -379,8 +379,8 @@ func TestIngestLedger_WritesEveryHotType(t *testing.T) { assert.Equal(t, first, seq) bms, err := db.Events().LookupKeys(context.Background(), []event.TermKey{term}) require.NoError(t, err) - require.NotNil(t, bms[0]) - assert.Equal(t, uint64(1), bms[0].GetCardinality()) + require.True(t, bms[0].Present()) + assert.Equal(t, uint64(1), bms[0].Cardinality()) } // TestIngestLedger_EventlessTxStillIndexesHash pins the post-merge txhash