fixes

Author: julienrbrt

block/components.go

@@ -178,13 +178,11 @@ func NewSyncComponents(
 		syncer.SetBlockSyncer(syncing.WithTracingBlockSyncer(syncer))
 	}
 
-	var execPruner pruner.ExecMetaPruner
-	if exec != nil {
-		if candidate, ok := exec.(pruner.ExecMetaPruner); ok {
-			execPruner = candidate
-		}
+	var execPruner pruner.ExecPruner
+	if p, ok := exec.(pruner.ExecPruner); ok {
+		execPruner = p
 	}
-	recoveryPruner := pruner.New(store, execPruner, config.Node.RecoveryHistoryDepth, pruner.DefaultPruneInterval, logger.With().Str("component", "Pruner").Logger())
+	pruner := pruner.New(logger, store, execPruner, config.Node)
 
 	// Create submitter for sync nodes (no signer, only DA inclusion processing)
 	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerDAHintAppender, dataDAHintAppender)
@@ -209,7 +207,7 @@ func NewSyncComponents(
 		Syncer:    syncer,
 		Submitter: submitter,
 		Cache:     cacheManager,
-		Pruner:    recoveryPruner,
+		Pruner:    pruner,
 		errorCh:   errorCh,
 	}, nil
 }
@@ -269,13 +267,11 @@ func NewAggregatorComponents(
 		executor.SetBlockProducer(executing.WithTracingBlockProducer(executor))
 	}
 
-	var execPruner pruner.ExecMetaPruner
-	if exec != nil {
-		if candidate, ok := exec.(pruner.ExecMetaPruner); ok {
-			execPruner = candidate
-		}
+	var execPruner pruner.ExecPruner
+	if p, ok := exec.(pruner.ExecPruner); ok {
+		execPruner = p
 	}
-	recoveryPruner := pruner.New(store, execPruner, config.Node.RecoveryHistoryDepth, pruner.DefaultPruneInterval, logger.With().Str("component", "Pruner").Logger())
+	pruner := pruner.New(logger, store, execPruner, config.Node)
 
 	reaper, err := reaping.NewReaper(
 		exec,
@@ -293,7 +289,7 @@ func NewAggregatorComponents(
 	if config.Node.BasedSequencer { // no submissions needed for bases sequencer
 		return &Components{
 			Executor: executor,
-			Pruner:   recoveryPruner,
+			Pruner:   pruner,
 			Reaper:   reaper,
 			Cache:    cacheManager,
 			errorCh:  errorCh,
@@ -320,7 +316,7 @@ func NewAggregatorComponents(
 
 	return &Components{
 		Executor:  executor,
-		Pruner:    recoveryPruner,
+		Pruner:    pruner,
 		Reaper:    reaper,
 		Submitter: submitter,
 		Cache:     cacheManager,

block/internal/pruner/pruner.go

@@ -9,18 +9,19 @@ import (
 	ds "github.com/ipfs/go-datastore"
 	"github.com/rs/zerolog"
 
+	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/store"
 )
 
 const (
-	DefaultPruneInterval = 15 * time.Minute
+	defaultPruneInterval = 15 * time.Minute
 	// maxPruneBatch limits how many heights we prune per cycle to bound work.
 	maxPruneBatch = uint64(1000)
 )
 
-// ExecMetaPruner removes execution metadata at a given height.
-type ExecMetaPruner interface {
-	PruneExecMeta(ctx context.Context, height uint64) error
+// ExecPruner removes execution metadata at a given height.
+type ExecPruner interface {
+	PruneExec(ctx context.Context, height uint64) error
 }
 
 type stateDeleter interface {
@@ -31,22 +32,24 @@ type stateDeleter interface {
 type Pruner struct {
 	store        store.Store
 	stateDeleter stateDeleter
-	execPruner   ExecMetaPruner
-	retention    uint64
-	interval     time.Duration
+	execPruner   ExecPruner
+	cfg          config.NodeConfig
 	logger       zerolog.Logger
 	lastPruned   uint64
 
+	// Lifecycle
+	ctx    context.Context
 	wg     sync.WaitGroup
 	cancel context.CancelFunc
 }
 
 // New creates a new Pruner instance.
-func New(store store.Store, execMetaPruner ExecMetaPruner, retention uint64, interval time.Duration, logger zerolog.Logger) *Pruner {
-	if interval <= 0 {
-		interval = DefaultPruneInterval
-	}
-
+func New(
+	logger zerolog.Logger,
+	store store.Store,
+	execPruner ExecPruner,
+	cfg config.NodeConfig,
+) *Pruner {
 	var deleter stateDeleter
 	if store != nil {
 		if sd, ok := store.(stateDeleter); ok {
@@ -57,75 +60,66 @@ func New(store store.Store, execMetaPruner ExecMetaPruner, retention uint64, int
 	return &Pruner{
 		store:        store,
 		stateDeleter: deleter,
-		execPruner:   execMetaPruner,
-		retention:    retention,
-		interval:     interval,
-		logger:       logger,
+		execPruner:   execPruner,
+		cfg:          cfg,
+		logger:       logger.With().Str("component", "prune").Logger(),
 	}
 }
 
 // Start begins the pruning loop.
 func (p *Pruner) Start(ctx context.Context) error {
-	if p.retention == 0 {
-		return nil
-	}
+	p.ctx, p.cancel = context.WithCancel(ctx)
 
-	loopCtx, cancel := context.WithCancel(ctx)
-	p.cancel = cancel
-
-	p.wg.Add(1)
-	go p.pruneLoop(loopCtx)
+	// Start pruner loop
+	p.wg.Go(p.pruneLoop)
 
+	p.logger.Info().Msg("pruner started")
 	return nil
 }
 
 // Stop stops the pruning loop.
 func (p *Pruner) Stop() error {
-	if p == nil || p.cancel == nil {
-		return nil
+	if p.cancel != nil {
+		p.cancel()
 	}
-
-	p.cancel()
 	p.wg.Wait()
+
+	p.logger.Info().Msg("pruner stopped")
 	return nil
 }
 
-func (p *Pruner) pruneLoop(ctx context.Context) {
-	defer p.wg.Done()
-	ticker := time.NewTicker(p.interval)
+func (p *Pruner) pruneLoop() {
+	ticker := time.NewTicker(defaultPruneInterval)
 	defer ticker.Stop()
 
-	if err := p.pruneOnce(ctx); err != nil {
-		p.logger.Error().Err(err).Msg("failed to prune recovery history")
-	}
-
 	for {
 		select {
 		case <-ticker.C:
-			if err := p.pruneOnce(ctx); err != nil {
+			if err := p.pruneRecoveryHistory(p.ctx, p.cfg.RecoveryHistoryDepth); err != nil {
 				p.logger.Error().Err(err).Msg("failed to prune recovery history")
 			}
-		case <-ctx.Done():
+
+			// TODO: add pruning of old blocks // https://github.com/evstack/ev-node/pull/2984
+		case <-p.ctx.Done():
 			return
 		}
 	}
 }
 
-func (p *Pruner) pruneOnce(ctx context.Context) error {
-	if p.retention == 0 || p.store == nil {
-		return nil
-	}
-
+// pruneRecoveryHistory prunes old state and execution metadata entries based on the configured retention depth.
+// It does not prunes old blocks, as those are handled by the pruning logic.
+// Pruning old state does not lose history but limit the ability to recover (replay or rollback) to the last HEAD-N blocks, where N is the retention depth.
+func (p *Pruner) pruneRecoveryHistory(ctx context.Context, retention uint64) error {
 	height, err := p.store.Height(ctx)
 	if err != nil {
 		return err
 	}
 
-	if height <= p.retention {
+	if height <= retention {
 		return nil
 	}
 
-	target := height - p.retention
+	target := height - retention
 	if target <= p.lastPruned {
 		return nil
 	}
@@ -143,7 +137,7 @@ func (p *Pruner) pruneOnce(ctx context.Context) error {
 			}
 		}
 		if p.execPruner != nil {
-			if err := p.execPruner.PruneExecMeta(ctx, h); err != nil && !errors.Is(err, ds.ErrNotFound) {
+			if err := p.execPruner.PruneExec(ctx, h); err != nil && !errors.Is(err, ds.ErrNotFound) {
 				return err
 			}
 		}

block/internal/pruner/pruner_test.go

@@ -3,13 +3,13 @@ package pruner
 import (
 	"context"
 	"testing"
-	"time"
 
 	ds "github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/require"
 
+	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/types"
 )
@@ -18,7 +18,7 @@ type execMetaAdapter struct {
 	existing map[uint64]struct{}
 }
 
-func (e *execMetaAdapter) PruneExecMeta(ctx context.Context, height uint64) error {
+func (e *execMetaAdapter) PruneExec(ctx context.Context, height uint64) error {
 	delete(e.existing, height)
 	return nil
 }
@@ -40,8 +40,8 @@ func TestPrunerPrunesRecoveryHistory(t *testing.T) {
 
 	execAdapter := &execMetaAdapter{existing: map[uint64]struct{}{1: {}, 2: {}, 3: {}}}
 
-	recoveryPruner := New(stateStore, execAdapter, 2, time.Minute, zerolog.Nop())
-	require.NoError(t, recoveryPruner.pruneOnce(ctx))
+	recoveryPruner := New(zerolog.Nop(), stateStore, execAdapter, config.NodeConfig{RecoveryHistoryDepth: 2})
+	require.NoError(t, recoveryPruner.pruneRecoveryHistory(ctx, recoveryPruner.cfg.RecoveryHistoryDepth))
 
 	_, err := stateStore.GetStateAtHeight(ctx, 1)
 	require.ErrorIs(t, err, ds.ErrNotFound)

pkg/store/store.go

@@ -30,14 +30,6 @@ func New(ds ds.Batching) Store {
 	}
 }
 
-// DeleteStateAtHeight removes the state entry at the given height.
-func (s *DefaultStore) DeleteStateAtHeight(ctx context.Context, height uint64) error {
-	if err := s.db.Delete(ctx, ds.NewKey(getStateAtHeightKey(height))); err != nil && !errors.Is(err, ds.ErrNotFound) {
-		return fmt.Errorf("failed to delete state at height %d: %w", height, err)
-	}
-	return nil
-}
-
 // Close safely closes underlying data storage, to ensure that data is actually saved.
 func (s *DefaultStore) Close() error {
 	return s.db.Close()
@@ -180,6 +172,14 @@ func (s *DefaultStore) GetStateAtHeight(ctx context.Context, height uint64) (typ
 	return state, nil
 }
 
+// DeleteStateAtHeight removes the state entry at the given height.
+func (s *DefaultStore) DeleteStateAtHeight(ctx context.Context, height uint64) error {
+	if err := s.db.Delete(ctx, ds.NewKey(getStateAtHeightKey(height))); err != nil && !errors.Is(err, ds.ErrNotFound) {
+		return fmt.Errorf("failed to delete state at height %d: %w", height, err)
+	}
+	return nil
+}
+
 // SetMetadata saves arbitrary value in the store.
 //
 // Metadata is separated from other data by using prefix in KV.