events: store small terms as delta postings and query them un-materialized

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 ─────────────────────────

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
 }

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
 
 // ──────────────────────────────────────────────────────────────────

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)
 		}
 	}