Merge branch 'main' into cian/benchmark-results-artifact

Author: chatton

CHANGELOG.md

@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - Avoid evicting yet to be processed heights [#3204](https://github.com/evstack/ev-node/pull/3204)
+- Bound Badger index cache memory to prevent growth with chain length [3209](https://github.com/evstack/ev-node/pull/3209)
 - Refetch latest da height instead of da height +1 when P2P is offline [#3201](https://github.com/evstack/ev-node/pull/3201)
 - Fix race on startup sync. [#3162](https://github.com/evstack/ev-node/pull/3162)
 - Strict raft state. [#3167](https://github.com/evstack/ev-node/pull/3167)

block/internal/syncing/syncer.go

@@ -517,6 +517,11 @@ func (s *Syncer) waitForGenesis() bool {
 }
 
 func (s *Syncer) PipeEvent(ctx context.Context, event common.DAHeightEvent) error {
+	// Avoid sending already seen events to channel (would have been skipped in processHeightEvent anyway)
+	if s.cache.IsHeaderSeen(event.Header.Hash().String()) {
+		return nil
+	}
+
 	select {
 	case s.heightInCh <- event:
 		return nil
@@ -537,6 +542,7 @@ func (s *Syncer) processHeightEvent(ctx context.Context, event *common.DAHeightE
 		Uint64("height", height).
 		Uint64("da_height", event.DaHeight).
 		Str("hash", headerHash).
+		Str("source", string(event.Source)).
 		Msg("processing height event")
 
 	currentHeight, err := s.store.Height(ctx)
@@ -547,7 +553,10 @@ func (s *Syncer) processHeightEvent(ctx context.Context, event *common.DAHeightE
 
 	// Skip if already processed
 	if height <= currentHeight || s.cache.IsHeaderSeen(headerHash) {
-		s.logger.Debug().Uint64("height", height).Msg("height already processed")
+		s.logger.Debug().
+			Uint64("height", height).
+			Str("source", string(event.Source)).
+			Msg("height already processed")
 		return
 	}
 
@@ -656,6 +665,7 @@ func (s *Syncer) processHeightEvent(ctx context.Context, event *common.DAHeightE
 		s.logger.Error().Err(err).
 			Uint64("event-height", event.Header.Height()).
 			Uint64("state-height", s.getLastState().LastBlockHeight).
+			Str("source", string(event.Source)).
 			Msg("failed to sync next block")
 		// If the error is not due to a validation error, re-store the event as pending
 		switch {

pkg/store/badger_options.go

@@ -6,11 +6,20 @@ import (
 	badger4 "github.com/ipfs/go-ds-badger4"
 )
 
+const (
+	// DefaultBadgerIndexCacheSize bounds Badger's table index and bloom filter
+	// cache to avoid growth with chain length.
+	DefaultBadgerIndexCacheSize int64 = 256 << 20 // 256 MiB
+)
+
 // BadgerOptions returns ev-node tuned Badger options for the node workload.
 // These defaults favor write throughput for append-heavy usage.
 func BadgerOptions() *badger4.Options {
 	opts := badger4.DefaultOptions
 
+	// Bound Badger-owned caches explicitly instead of inheriting a large block
+	// cache and an unbounded index cache from the upstream defaults.
+	opts.Options = opts.WithIndexCacheSize(DefaultBadgerIndexCacheSize)
 	// Disable conflict detection to reduce write overhead; ev-node does not rely
 	// on Badger's multi-writer conflict checks for correctness.
 	opts.Options = opts.WithDetectConflicts(false)