ledger: compress each hot ledger alongside the walk, on two zstd threads

cmd/stellar-rpc/internal/rpcv2/ingest/service.go

@@ -62,17 +62,19 @@ func NewHotService(db *hotchunk.DB, windows *feewindow.FeeWindows, sink MetricSi
 // ledger and fails run() before the loop starts — the daemon wedges rather
 // than serve fee stats that silently miss a ledger.
 func (s *HotService) Ingest(seq uint32, lcmView xdr.LedgerCloseMetaView) (int64, error) {
+	pending := s.db.StartCompress(seq, lcmView)
 	walkStart := time.Now()
 	txParts, err := sdkingest.ExtractLedgerTxParts(lcmView)
 	walkDur := time.Since(walkStart)
 	if err != nil {
+		pending.Discard()
 		// The walk failed before any batch opened: the extract phase is the only
 		// one that ran, mirroring hotchunk's own pre-batch failures.
 		s.sink.HotPhase(hotchunk.PhaseExtract, walkDur, 0, err)
 		return 0, fmt.Errorf("extract ledger tx parts seq %d: %w", seq, err)
 	}
 
-	rep, err := s.db.IngestLedger(seq, lcmView, txParts)
+	rep, err := s.db.IngestLedger(seq, lcmView, txParts, pending)
 	rep.Phases[hotchunk.PhaseExtract].Dur += walkDur
 
 	last := hotchunk.NumPhases - 1

cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go

@@ -167,7 +167,8 @@ func IngestLedger(t testing.TB, db *hotchunk.DB, seq uint32, raw []byte) {
 	t.Helper()
 	txParts, err := sdkingest.ExtractLedgerTxParts(xdr.LedgerCloseMetaView(raw))
 	require.NoError(t, err)
-	_, err = db.IngestLedger(seq, xdr.LedgerCloseMetaView(raw), txParts)
+	view := xdr.LedgerCloseMetaView(raw)
+	_, err = db.IngestLedger(seq, view, txParts, db.StartCompress(seq, view))
 	require.NoError(t, err)
 }
 

cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk.go

@@ -359,6 +359,12 @@ type LedgerReport struct {
 	CloseTime int64
 }
 
+// StartCompress starts compressing the ledger IngestLedger will write. The
+// returned pending must be passed to IngestLedger, or discarded.
+func (d *DB) StartCompress(seq uint32, lcmView xdr.LedgerCloseMetaView) *ledger.PendingCompression {
+	return d.ledger.StartCompress(ledger.Entry{Seq: seq, Bytes: []byte(lcmView)})
+}
+
 // IngestLedger commits ONE ledger as a SINGLE atomic synced WriteBatch across all
 // hot CFs (decision (a)): queue ledgers, txhash, and events rows into one
 // BatchWriter, commit once, and only then apply the txhash window and events
@@ -370,13 +376,18 @@ type LedgerReport struct {
 // serving state. lcmView is still needed for the raw-ledgers write and the
 // close time.
 //
+// pending is lcmView's compression, which the caller starts with StartCompress
+// before its walk so the two overlap. IngestLedger joins it on every path.
+//
 // lcmView is a borrowed zero-copy view and txParts aliases it; every extractor
 // copies what it retains, so neither need outlive this call. Store.Batch's
 // lifecycle RLock + checkOpen is the authoritative closed-store guard, so there
 // is no separate pre-check here.
 func (d *DB) IngestLedger(
 	seq uint32, lcmView xdr.LedgerCloseMetaView, txParts []sdkingest.LedgerTxParts,
+	pending *ledger.PendingCompression,
 ) (LedgerReport, error) {
+	defer pending.Discard()
 	var rep LedgerReport
 
 	// A read-only (ledgers-only) DB has no events facade to assign event IDs, and
@@ -440,14 +451,6 @@ func (d *DB) IngestLedger(
 	failed := PhaseCommit
 	batchStart := time.Now()
 	cerr := d.store.Batch(func(b *rocksdb.BatchWriter) error {
-		ls := time.Now()
-		if err := d.ledger.AddLedgerToBatch(b, ledger.Entry{Seq: seq, Bytes: []byte(lcmView)}); err != nil {
-			rep.Phases[PhaseLedgers].Dur = time.Since(ls)
-			failed = PhaseLedgers
-			return fmt.Errorf("queue ledger seq %d: %w", seq, err)
-		}
-		rep.Phases[PhaseLedgers].Dur = time.Since(ls)
-
 		ts := time.Now()
 		applyTxhash = d.txhash.AddRowToBatch(b, seq, txRow)
 		rep.Phases[PhaseTxhash].Dur = time.Since(ts)
@@ -461,6 +464,16 @@ func (d *DB) IngestLedger(
 		}
 		rep.Phases[PhaseEvents].Dur = time.Since(es)
 		applyEvents = apply
+
+		// The ledger's compression is joined last, so it has the longest to
+		// finish; PhaseLedgers is the wait for it plus the Put.
+		ls := time.Now()
+		if err := d.ledger.AddPendingToBatch(b, pending); err != nil {
+			rep.Phases[PhaseLedgers].Dur = time.Since(ls)
+			failed = PhaseLedgers
+			return fmt.Errorf("queue ledger seq %d: %w", seq, err)
+		}
+		rep.Phases[PhaseLedgers].Dur = time.Since(ls)
 		return nil
 	})
 	// Commit is the whole Batch call minus the three queue steps: the RocksDB write

cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk_test.go

@@ -56,7 +56,8 @@ func ingestRaw(t *testing.T, db *DB, seq uint32, raw []byte) (LedgerReport, erro
 	t.Helper()
 	txParts, err := sdkingest.ExtractLedgerTxParts(xdr.LedgerCloseMetaView(raw))
 	require.NoError(t, err)
-	return db.IngestLedger(seq, xdr.LedgerCloseMetaView(raw), txParts)
+	view := xdr.LedgerCloseMetaView(raw)
+	return db.IngestLedger(seq, view, txParts, db.StartCompress(seq, view))
 }
 
 // openTestDB opens a fresh hot DB bound to chunk 0 (every test uses chunk 0).

cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store.go

@@ -9,6 +9,7 @@ import (
 	"iter"
 	"slices"
 	"sync"
+	"sync/atomic"
 
 	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/rocksdb"
 	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/stores"
@@ -36,22 +37,27 @@ type Entry struct {
 // before freezing; it does not itself range-check writes (the driver's drain loop
 // already validates every sequence against the chunk).
 //
-// Concurrency: all methods are safe for concurrent use, including use alongside
-// the caller-owned rocksdb.Store.Close. A read/write racing Close either completes
-// first or observes the closed store and returns stores.ErrStoreClosed. HotStore
-// adds no unguarded state of its own — the compressor pool and decompressor are
-// both concurrent-safe.
+// Concurrency: the read methods are safe for concurrent use, including use
+// alongside the caller-owned rocksdb.Store.Close. A read/write racing Close either
+// completes first or observes the closed store and returns stores.ErrStoreClosed.
+// The writes compress one ledger at a time: StartCompress panics while another
+// compression is in flight.
 type HotStore struct {
 	store *rocksdb.Store
 	dec   *zstd.Decompressor
-	// compPool — per-store pool of zstd.Compressors; each concurrent
-	// AddLedgerToBatch borrows one for its Encode call.
-	compPool sync.Pool
+	// comp compresses every ledger into buf, which keeps the largest
+	// compressed ledger's capacity for the next one.
+	comp     *zstd.Compressor
+	buf      []byte
+	inFlight atomic.Bool
 	// scratch — decode buffers WithLedger lends. Pooled as *[]byte so a buffer
 	// the decode had to grow goes back in place of the one that was lent.
 	scratch sync.Pool
 }
 
+// zstdWorkers is how many libzstd threads compress each ledger.
+const zstdWorkers = 2
+
 // maxPooledLedgerBytes is the largest decode buffer this store keeps. Capacity
 // only ratchets upward and sync.Pool accepts whatever it is given, so without a
 // ceiling N concurrent borrows can park N outsized buffers for the store's life.
@@ -66,32 +72,81 @@ func NewWithStore(store *rocksdb.Store) *HotStore {
 	return &HotStore{
 		store: store,
 		dec:   zstd.NewDecompressor(),
-		compPool: sync.Pool{
-			New: func() any { return zstd.NewCompressor() },
-		},
+		comp:  zstd.NewCompressor(zstd.WithWorkers(zstdWorkers)),
 		scratch: sync.Pool{
 			New: func() any { return new([]byte) },
 		},
 	}
 }
 
-// AddLedgerToBatch compresses one ledger and queues its Put into b on LedgersCF
-// — the building block hotchunk uses to fold the ledger write into the one
-// shared per-ledger WriteBatch (decision (a)). Does not commit (caller owns the
-// batch). Compresses into a fresh buffer BatchWriter.Put copies, so e.Bytes need
-// not outlive this call. The caller runs inside Store.Batch, whose lifecycle
-// RLock + checkOpen is the authoritative closed-store guard, so this adds none.
-func (h *HotStore) AddLedgerToBatch(b *rocksdb.BatchWriter, e Entry) error {
-	c, _ := h.compPool.Get().(*zstd.Compressor)
-	defer h.compPool.Put(c)
-	compressed, err := c.Encode(nil, e.Bytes)
-	if err != nil {
-		return err
+// PendingCompression is a ledger StartCompress is compressing in the
+// background. AddPendingToBatch or Discard must join it, and both block until
+// the compression is done with the ledger's bytes. Discard after
+// AddPendingToBatch does nothing, so it can be deferred.
+type PendingCompression struct {
+	h          *HotStore
+	seq        uint32
+	done       chan struct{}
+	compressed []byte
+	err        error
+	joined     bool
+}
+
+// StartCompress compresses e.Bytes on its own goroutine, so the caller can
+// read the ledger meanwhile. e.Bytes must stay valid until the pending is
+// joined.
+func (h *HotStore) StartCompress(e Entry) *PendingCompression {
+	if !h.inFlight.CompareAndSwap(false, true) {
+		panic("ledger: StartCompress while another compression is in flight")
+	}
+	p := &PendingCompression{h: h, seq: e.Seq, done: make(chan struct{})}
+	dst := h.buf[:0]
+	go func() {
+		defer close(p.done)
+		p.compressed, p.err = h.comp.Encode(dst, e.Bytes)
+	}()
+	return p
+}
+
+// AddPendingToBatch joins p and queues its compressed ledger into b on
+// LedgersCF, the ledger's part of the one shared per-ledger WriteBatch
+// (decision (a)). Does not commit (caller owns the batch). The caller runs
+// inside Store.Batch, whose lifecycle RLock + checkOpen is the authoritative
+// closed-store guard, so this adds none.
+func (h *HotStore) AddPendingToBatch(b *rocksdb.BatchWriter, p *PendingCompression) error {
+	<-p.done
+	defer p.release()
+	if p.err != nil {
+		return p.err
 	}
-	b.Put(LedgersCF, rocksdb.EncodeUint32(e.Seq), compressed)
+	b.Put(LedgersCF, rocksdb.EncodeUint32(p.seq), p.compressed)
 	return nil
 }
 
+// Discard joins p unless AddPendingToBatch has, dropping its result.
+func (p *PendingCompression) Discard() {
+	if !p.joined {
+		<-p.done
+		p.release()
+	}
+}
+
+// release returns the buffer, which BatchWriter.Put has copied, for the next
+// ledger.
+func (p *PendingCompression) release() {
+	p.joined = true
+	if p.err == nil {
+		p.h.buf = p.compressed[:cap(p.compressed)]
+	}
+	p.h.inFlight.Store(false)
+}
+
+// AddLedgerToBatch compresses one ledger and queues its Put into b, as
+// StartCompress and AddPendingToBatch do.
+func (h *HotStore) AddLedgerToBatch(b *rocksdb.BatchWriter, e Entry) error {
+	return h.AddPendingToBatch(b, h.StartCompress(e))
+}
+
 // WithLedger calls fn with seq's decoded bytes; see query.LedgerReader for the
 // loan rule. The buffer returns to the store's pool as fn returns.
 func (h *HotStore) WithLedger(seq uint32, fn func(raw []byte) error) error {

cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"path/filepath"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -326,13 +327,14 @@ func TestHotStore_ConcurrentOpsAndCloseRaceFree(t *testing.T) {
 
 	var wg sync.WaitGroup
 	var stop atomic.Bool
-	const workers = 4
-	for w := range workers {
-		wg.Go(func() {
-			for i := uint32(0); !stop.Load(); i++ {
-				_ = addLedgers(h, Entry{Seq: uint32(w)*1_000_000 + i, Bytes: []byte("v")})
-			}
-		})
+	// One writer, as the writes require; the reads are concurrent.
+	wg.Go(func() {
+		for i := uint32(0); !stop.Load(); i++ {
+			_ = addLedgers(h, Entry{Seq: 1_000_000 + i, Bytes: []byte("v")})
+		}
+	})
+	const readers = 4
+	for range readers {
 		wg.Go(func() {
 			for i := uint32(0); !stop.Load(); i++ {
 				_, _ = readLedgerRaw(h, i%50)
@@ -544,3 +546,31 @@ func readLedgerRaw(r interface {
 	})
 	return out, err
 }
+
+// A second compression in flight panics; joining the first releases the
+// store for the next.
+func TestStartCompress_OneInFlight(t *testing.T) {
+	h := NewWithStore(nil) // compressing never touches the store
+	first := h.StartCompress(Entry{Seq: 1, Bytes: []byte("x")})
+	require.Panics(t, func() { h.StartCompress(Entry{Seq: 2, Bytes: []byte("x")}) })
+	first.Discard()
+	h.StartCompress(Entry{Seq: 3, Bytes: []byte("x")}).Discard()
+}
+
+// Every ledger is compressed into the one buffer the store keeps, across
+// garbage collections.
+func TestStartCompress_ReusesItsBuffer(t *testing.T) {
+	h := NewWithStore(nil)
+	payload := bytes.Repeat([]byte("ledger bytes "), 1<<16)
+	h.StartCompress(Entry{Seq: 1, Bytes: payload}).Discard()
+
+	var before, after runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&before)
+	for i := range uint32(20) {
+		runtime.GC()
+		h.StartCompress(Entry{Seq: i + 2, Bytes: payload}).Discard()
+	}
+	runtime.ReadMemStats(&after)
+	require.Less(t, after.TotalAlloc-before.TotalAlloc, uint64(len(payload)))
+}

cmd/stellar-rpc/internal/rpcv2/zstd/zstd.go

@@ -62,6 +62,7 @@ type CompressorOption func(*compressorConfig)
 
 type compressorConfig struct {
 	checksum bool
+	workers  int
 }
 
 // WithoutChecksum disables the zstd content checksum.
@@ -70,6 +71,13 @@ func WithoutChecksum() CompressorOption {
 	return func(cfg *compressorConfig) { cfg.checksum = false }
 }
 
+// WithWorkers compresses on n libzstd threads. The output is one standard
+// frame, though not the bytes a single-threaded encode of the same input
+// gives. NewCompressor panics when libzstd was built without multithreading.
+func WithWorkers(n int) CompressorOption {
+	return func(cfg *compressorConfig) { cfg.workers = n }
+}
+
 // NewCompressor creates a new Compressor. By default content checksums are enabled.
 func NewCompressor(opts ...CompressorOption) *Compressor {
 	checkVersion()
@@ -93,6 +101,12 @@ func NewCompressor(opts ...CompressorOption) *Compressor {
 		C.ZSTD_freeCCtx(ctx)
 		panic("zstd: set checksum flag: " + C.GoString(C.ZSTD_getErrorName(rc)))
 	}
+	if cfg.workers > 0 {
+		if rc := C.ZSTD_CCtx_setParameter(ctx, C.ZSTD_c_nbWorkers, C.int(cfg.workers)); C.ZSTD_isError(rc) != 0 {
+			C.ZSTD_freeCCtx(ctx)
+			panic("zstd: set workers: " + C.GoString(C.ZSTD_getErrorName(rc)))
+		}
+	}
 	c := &Compressor{ctx: ctx}
 	// Safety net for C memory: when sync.Pool evicts items, there's no cleanup
 	// callback, so the finalizer ensures the C context is eventually freed.