Add eventstore: per-Chunk hot + cold event storage for full-history - #740

After the RocksDB batch commits durably (line 348), mirror.AddTo and offsets.Append are called with error returns that don't surface meaningfully. Both errors are unreachable under the function's preconditions (AddTo only fails with ErrClosed which the leading check rules out; Append only fails on sequence mismatch which the validation at line 307 already enforced). But typed-fallible code that can't actually fail forces every reader to reason about a recovery path that doesn't exist — and if either ever did fail post-batch, we'd silently violate the atomicity the type docstring at lines 98-101 advertises.

Replace the post-batch error returns with panic + rationale comment:

// Phase 3: apply the committed batch to the in-memory mirrors.
//
// mirror.AddTo and offsets.Append are typed as fallible but cannot
// fail here:
//   - mirror.AddTo errors only with ErrClosed; we hold h.mu and
//     closed is gated by the leading check.
//   - offsets.Append errors only on out-of-order ledger sequence,
//     which the validation at the top of this function already
//     enforced.
//
// Phase 2 has already committed to disk, so a non-nil return here
// would leave on-disk and in-memory state diverged with no clean
// recovery. Panic so a future regression surfaces loudly instead
// of silently corrupting the hot store.
for i, keys := range termKeys {
    eventID := startID + uint32(i)
    for _, key := range keys {
        if err := h.mirror.AddTo(key, eventID); err != nil {
            panic(fmt.Sprintf("events: mirror.AddTo invariant violation, chunk %s ledger %d: %v",
                h.chunkID, ledgerSeq, err))
        }
    }
}
if err := h.offsets.Append(ledgerSeq, uint32(len(payloads))); err != nil {
    panic(fmt.Sprintf("events: offsets.Append invariant violation, chunk %s ledger %d: %v",
        h.chunkID, ledgerSeq, err))
}
return nil

Update the type-level docstring to reflect the panic contract. BitmapIndex.AddTo and LedgerOffsets.Append stay typed-fallible (they may be called from contexts where errors are recoverable); this is a localized policy at the IngestLedgerEvents call site.

Walked through this with @tamirms — the panic recommendation here is overstated. The actual fix is to make the operation idempotent.

The RocksDB batch (phase 2) is already idempotent (same keys, same values, overwrite-on-Put). offsets.Append can't actually fail at this call site (its only failure mode is sequence mismatch, which the leading validation rules out). The only thing that prevents IngestLedgerEvents from being fully retry-safe is memBitmaps.AddTo in list mode — the te.ids = append(te.ids, eventIDs...) at membitmaps.go:109 accumulates duplicates if the eventID was already added.

Key observation: eventIDs are added in monotonically increasing order per term. IngestLedgerEvents assigns eventID = startID + i and iterates i = 0, 1, 2...; warmup iterates events_index in byte-lex order which is numeric order for the BE eventID suffix. So te.ids is always sorted, and any retry duplicate is <= the current last entry (not necessarily equal — it could be from the successful prefix of a partially-failed attempt).

O(1) dedupe:

if te.bm != nil {
    te.bm.AddMany(eventIDs) // already set-semantics; idempotent
} else {
    for _, id := range eventIDs {
        if len(te.ids) > 0 && te.ids[len(te.ids)-1] >= id {
            continue // already in the sorted prefix — retry duplicate
        }
        te.ids = append(te.ids, id)
    }
    if len(te.ids) >= promotionThreshold {
        te.bm = roaring.New()
        te.bm.AddMany(te.ids)
        te.ids = nil
    }
}

The >= is important — not just ==. Trace: if a previous IngestLedgerEvents attempt failed at i=3 after term T had already been added for i=0 and i=1, the retry comes in at i=0 with id = X+0 while te_T.ids = [X+0, X+1] and last = X+1. The == check would treat X+0 as new (because X+0 != X+1) and append, corrupting the list. >= correctly recognizes it as part of the already-added prefix and skips.

With that change:

• IngestLedgerEvents becomes fully retry-safe regardless of where phase 3 fails.
• Phase 3 keeps error returns (no panic).
• The atomicity docstring (lines 98-101) becomes honest — disk and mirror converge to the same state regardless of partial failure.
• Document BitmapIndex.AddTo contract: "AddTo is idempotent. Callers must add eventIDs in monotonically increasing order per term; the same (key, eventID) pair has the same effect added once or many times."

Withdrawing the panic recommendation above in favor of this.

Applied. Split the rationale across the two panics — mirror.AddTo is reachable only via orchestrator-coordination bug (mirror closed mid-ingest); offsets.Append is reachable only via a future refactor breaking the up-front validation. Function docstring grew a "Post-batch atomicity" paragraph.

Sorry — re-reading this thread, the dedup-in-AddTo follow-up was buried as a self-reply on the original panic recommendation, which made it easy to miss. The "Applied" tag here covered splitting the panic rationale (taking the original panic recommendation), but the dedup-in-AddTo proposal that I posted as a follow-up didn't make it across. Re-raising it now because it's the structural fix, and the current rationale comments are already stale:

• The mirror.AddTo panic comment says it's reachable only "if the freeze orchestrator closed the mirror while ingest was still running" — but BitmapIndex.Close was deleted in commit 331c341f. AddTo no longer has any failure mode, so its error return is vestigial and the panic guards a path that no longer exists.
• The offsets.Append panic is "defense in depth" for a hypothetical future refactor breaking phase 1.

Both panics exist as workarounds for AddTo not being idempotent. Idempotent AddTo removes the need for either:

if te.bm != nil {
    te.bm.AddMany(eventIDs)  // already set-semantics
} else {
    for _, id := range eventIDs {
        if len(te.ids) > 0 && te.ids[len(te.ids)-1] >= id {
            continue  // already in the sorted prefix — retry duplicate
        }
        te.ids = append(te.ids, id)
    }
    if len(te.ids) >= promotionThreshold {
        te.bm = roaring.New()
        te.bm.AddMany(te.ids)
        te.ids = nil
    }
}

With that:

• Phase 3 has no failure modes. Both panics go away.
• AddTo can drop its error return entirely.
• The atomicity contract becomes honest end-to-end — IngestLedgerEvents is fully retry-safe regardless of where partial failure happens, no orchestrator-coordination caveats needed.
• BitmapIndex.AddTo docstring states the contract: "AddTo is idempotent. Callers must add eventIDs in monotonically increasing order per term; the same (key, eventID) pair has the same effect added once or many times."

The structural fix is at the AddTo layer, not at the IngestLedgerEvents layer or the panic-vs-error question.

FetchEvents iterates eventIDs serially (ReadItem per ID on cold, Get per ID on hot). For typical queries (coordinator computes bitmap intersection, hundreds-to-thousands of IDs per chunk), this pays per-call overhead per ID.

Three coupled changes:

1. Change the Reader.FetchEvents interface signature to materialize a slice, accept context, and document a sorted-input requirement:

   FetchEvents(ctx context.Context, eventIDs []uint32) ([]events.Payload, error)

   Slice instead of iter.Seq2: we materialize internally anyway. ctx: lets the parallel I/O honor cancellation. Add to the docstring: "eventIDs must be sorted ascending with no duplicates. Coordinators iterating a bitmap intersection naturally satisfy this. Behavior is undefined otherwise."

2. ColdReader.FetchEvents uses c.events.ReadItems(ctx, positions, func(idx int, data []byte) error { return results[idx].Unmarshal(data) }). The packfile already supports parallel scattered reads via ReadItems; we just need to use it.

3. HotStore.FetchEvents matches the signature. The actual parallelism on the hot side comes via BatchedMultiGetCF — see the separate comment on pkg/rocksdb/rocksdb.go.

All keeps its iter.Seq2 shape — naturally streaming from a scan, no benefit to materializing.

Note this fix only delivers parallel reads if packfile.ReaderOptions.Concurrency is also set — see the comment on cold_reader.go below.

Applied (commit cec8ff82).

Interface change:

FetchEvents(ctx context.Context, eventIDs []uint32) ([]events.Payload, error)

Docstring states the sorted-ascending-no-duplicates precondition, calls out that roaring.Bitmap iteration satisfies it for free, and documents the asymmetric enforcement (cold returns wrapped packfile.ErrPositionsUnsorted; hot reads in given order — neither relied upon).

Cold side: delegates to packfile.ReadItems(ctx, positions, fn). The fn(idx, data) callback writes into results[idx], so concurrent workers produce ordered output by construction. Parallel fan-out kicks in when ColdReaderOptions.Concurrency > 1 (knob landed in batch 3).

Hot side: signature matches but the batching part (your separate comment #13) is in this commit too — details in the reply there.

All() shape: kept as iter.Seq2 per your note.

One thing to add: the applied docstring acknowledges the asymmetry as "hot reads in given order — neither relied upon", but BatchMultiGet is called with sortedInput=true (rocksdb.go:204-210), which tells RocksDB's BatchedMultiGetCF that the input is sorted. With unsorted input + sortedInput=true, BatchedMultiGetCF's behavior is implementation-defined — the hot side does silently rely on the precondition, just without the cold side's explicit error.

A cheap O(N) check before the encode loop makes both sides enforce the same contract:

for i := 1; i < len(eventIDs); i++ {
    if eventIDs[i] <= eventIDs[i-1] {
        return nil, fmt.Errorf("events: FetchEvents requires sorted ascending no-dup eventIDs (chunk %s)", h.chunkID)
    }
}

For symmetry with the cold side's wrapped packfile.ErrPositionsUnsorted, we could also define a package-level ErrUnsortedEventIDs and return that wrapped so callers can errors.Is against either path.

Also worth revising the implementation comment at hot_store.go line ~237 — "builds a sorted [][]byte of encoded eventID keys" is misleading since the slice isn't actively sorted; it's sorted post-validation by the caller's precondition.

events_offsets stores cumulative counts on disk and warmupOffsets deltas them back at startup. The on-disk format and the in-memory API (LedgerOffsets.Append(ledger, eventCount)) disagree on shape, requiring the delta arithmetic at warmup time.

Switch to per-ledger deltas on disk. Warmup becomes a direct offsets.Append(ledger, eventCount) — no delta arithmetic, the on-disk and in-memory shapes match. No correctness impact.

Applied. On-disk shape is now per-ledger event count (not cumulative). Warmup drops the delta arithmetic; the on-disk and in-memory shapes match. endID local var dropped (unused).

One leftover from this refactor: encodeOffsetValue(cumulative uint32) at hot_store.go:558 — the parameter name is stale from the pre-applied cumulative-on-disk shape. The caller now passes eventCount := uint32(len(payloads)) per-ledger, and the comment at line 407 explicitly says "per-ledger event count, not cumulative." The function name and parameter name still encode the old shape.

Minimum: rename cumulative → eventCount so the parameter matches what it represents.

Could also rename encodeOffsetValue → encodeLedgerEventCount for symmetry with the post-applied semantics (and the same on the decode side at warmup). The current encodeOffsetKey / encodeOffsetValue pair reads as "key/value of the offsets CF" but the value isn't an offset anymore.

Single-key Lookup forces multi-term queries to either serialize lookups or have the caller wrap each in its own goroutine. For a query that intersects N terms in a chunk (the standard pattern — contractID + topic filters), this means N MPHF queries + N separate packfile.ReadItem calls per chunk.

A batch API can do better by exploiting the existing primitives:

LookupKeys(ctx context.Context, keys []events.TermKey) ([]*roaring.Bitmap, error)

Implementation sketch for ColdReader:

1. MPHF-resolve all keys to slots in one in-memory pass.
2. Sort by slot, dedupe (multiple keys may collide to the same MPHF slot in pathological cases).
3. Single c.index.ReadItems(ctx, sortedSlots, fn) call — same primitive we use for FetchEvents. The packfile layer coalesces adjacent slot reads into single ReadAt calls and fans out across the worker count configured via ColdReaderOptions.Concurrency.
4. Scatter the resulting bitmaps back to caller-input positions.

HotStore's implementation is trivially N in-memory mirror.Get calls (no I/O to batch).

Wins:

• One ReadItems syscall path instead of N ReadItem calls — same shape we already chose for FetchEvents.
• Sort-by-slot enables coalescing of adjacent records into a single read.
• Caller doesn't have to write goroutine fan-out at every multi-term call site.
• Result positions align with input — caller's intersection pipeline doesn't need to track ordering.

Suggested signature on Reader:

type Reader interface {
    // ... existing methods ...

    // LookupKeys returns bitmaps for each key, aligned positionally with the
    // input slice. result[i] is nil if keys[i] has no matching events in this
    // chunk; the function does not return ErrTermNotFound for individual misses.
    LookupKeys(ctx context.Context, keys []events.TermKey) ([]*roaring.Bitmap, error)
}

openHotChunk opens the per-chunk RocksDB without Tuning, so the three CFs inherit applyPinnedCFOptions — including SetCompression(NoCompression) at rocksdb.go:530 — and the default BBTO (16 KiB block) via applySharedTableOptions. Two related gaps:

(1) Compression on DataCF

DataCF holds XDR-encoded event payloads — significant compression headroom (repeated field tags, padding, similar structures across events). IndexCF (20-byte hash keys, empty values) and OffsetsCF (4-byte rows) have nothing to compress; leave them as NoCompression.

Recommendation: enable ZSTDCompression on DataCF. Per-block decompression CPU on cache-miss reads is real but well below the per-IOP latency on EBS-class storage.

(2) Block size per CF

The shared BBTO uses RocksDB's default 16 KiB block for all CFs. Two adjustments fit the eventstore CFs:

• DataCF: bump to 32 KiB. Larger blocks give zstd more context per compression unit and amortize per-block headers; batch-fetch shapes (FetchEvents) read multiple events per block.
• IndexCF: drop to 4 KiB. Random Lookups touch dense small-key blocks one at a time; smaller blocks let RocksDB's block cache hold more distinct blocks, improving hit ratio for the random-access pattern.
• OffsetsCF: same as IndexCF.

Structural cause: the wrapper pins compression in applyPinnedCFOptions and uses a shared BBTO with no per-CF override. To enable the per-CF policies above, Tuning would need per-CF override hooks for Compression and BlockSize. The existing per-CF knobs (write buffer, level0 thresholds) suggest the same shape would fit.