diff --git a/cmd/stellar-rpc/internal/rpcv2/ingest/service.go b/cmd/stellar-rpc/internal/rpcv2/ingest/service.go index 1a2ab23f4..c12f7f5dd 100644 --- a/cmd/stellar-rpc/internal/rpcv2/ingest/service.go +++ b/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 diff --git a/cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go b/cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go index 29d532c77..0ec59e3ff 100644 --- a/cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go +++ b/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) } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk.go b/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk.go index 2451e0c45..bdfecba5a 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk.go +++ b/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 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 994bd8c04..d3e9e3591 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk/hotchunk_test.go +++ b/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). diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store.go b/cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store.go index 81593fc0f..953fc0255 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store.go +++ b/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 { diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store_test.go index 5961aee9d..6379d6e59 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/ledger/hot_store_test.go +++ b/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))) +} diff --git a/cmd/stellar-rpc/internal/rpcv2/zstd/zstd.go b/cmd/stellar-rpc/internal/rpcv2/zstd/zstd.go index 367b3cdc2..7931fbc08 100644 --- a/cmd/stellar-rpc/internal/rpcv2/zstd/zstd.go +++ b/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.