diff --git a/block/components.go b/block/components.go
index 5eb5855d28..b9cc7766ed 100644
--- a/block/components.go
+++ b/block/components.go
@@ -186,7 +186,7 @@ func NewSyncComponents(
 	if p, ok := exec.(coreexecutor.ExecPruner); ok {
 		execPruner = p
 	}
-	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration, config.DA.Address)
+	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration)
 
 	var submitter *submitting.Submitter
 	if daClient != nil {
@@ -278,7 +278,7 @@ func newAggregatorComponents(
 	if p, ok := exec.(coreexecutor.ExecPruner); ok {
 		execPruner = p
 	}
-	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration, config.DA.Address)
+	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration)
 
 	reaper, err := reaping.NewReaper(
 		exec,
diff --git a/block/internal/pruner/pruner.go b/block/internal/pruner/pruner.go
index e525ca8a6a..ee11e19f9e 100644
--- a/block/internal/pruner/pruner.go
+++ b/block/internal/pruner/pruner.go
@@ -16,13 +16,12 @@ import (
 	"github.com/evstack/ev-node/pkg/store"
 )
 
-// Pruner periodically removes old state and execution metadata entries.
+// Pruner periodically removes old blocks, state snapshots, and execution metadata.
 type Pruner struct {
 	store      store.Store
 	execPruner coreexecutor.ExecPruner
 	cfg        config.PruningConfig
 	blockTime  time.Duration
-	daEnabled  bool
 	logger     zerolog.Logger
 
 	// Lifecycle
@@ -38,14 +37,12 @@ func New(
 	execPruner coreexecutor.ExecPruner,
 	cfg config.PruningConfig,
 	blockTime time.Duration,
-	daAddress string,
 ) *Pruner {
 	return &Pruner{
 		store:      store,
 		execPruner: execPruner,
 		cfg:        cfg,
 		blockTime:  blockTime,
-		daEnabled:  daAddress != "", // DA is enabled if address is provided
 		logger:     logger.With().Str("component", "pruner").Logger(),
 	}
 }
@@ -108,75 +105,71 @@ func (p *Pruner) pruneLoop() {
 	}
 }
 
-// pruneBlocks prunes blocks and their metadatas.
+// pruneBlocks prunes blocks and state snapshots outside local retention.
 func (p *Pruner) pruneBlocks() error {
 	storeHeight, err := p.store.Height(p.ctx)
 	if err != nil {
 		return fmt.Errorf("failed to get store height for pruning: %w", err)
 	}
 
-	upperBound := storeHeight
-
-	// If DA is enabled, only prune blocks that are DA included
-	if p.daEnabled {
-		var currentDAIncluded uint64
-		currentDAIncludedBz, err := p.store.GetMetadata(p.ctx, store.DAIncludedHeightKey)
-		if err == nil && len(currentDAIncludedBz) == 8 {
-			currentDAIncluded = binary.LittleEndian.Uint64(currentDAIncludedBz)
-		} else {
-			p.logger.Debug().Msg("skipping pruning: DA is enabled but DA included height is not available yet")
-			return nil
-		}
-
-		// Never prune blocks that are not DA included
-		upperBound = min(storeHeight, currentDAIncluded)
-	}
-
-	if upperBound <= p.cfg.KeepRecent {
-		// Not enough fully included blocks to prune
+	// Retention follows the locally committed head. DA inclusion can remain
+	// stalled when historical DA heights are no longer served, and must not
+	// prevent operators from reclaiming local storage.
+	if storeHeight <= p.cfg.KeepRecent {
 		return nil
 	}
+	targetHeight := storeHeight - p.cfg.KeepRecent
 
-	targetHeight := upperBound - p.cfg.KeepRecent
-
-	// Get the last pruned height to determine batch size
-	lastPruned, err := p.getLastPrunedBlockHeight(p.ctx)
+	lastBlock, err := p.getLastPrunedBlockHeight(p.ctx)
 	if err != nil {
 		return fmt.Errorf("failed to get last pruned block height: %w", err)
 	}
-
-	// prune in batches to avoid overwhelming the system
-	batchSize := p.calculateBatchSize()
-	batchEnd := min(lastPruned+batchSize, targetHeight)
-
-	if err := p.store.PruneBlocks(p.ctx, batchEnd); err != nil {
-		p.logger.Error().Err(err).Uint64("target_height", batchEnd).Msg("failed to prune old block data")
-		return err
+	lastState, err := p.getLastPrunedStateHeight(p.ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get last pruned state height: %w", err)
 	}
 
-	if p.execPruner != nil {
-		if err := p.execPruner.PruneExec(p.ctx, batchEnd); err != nil && !errors.Is(err, ds.ErrNotFound) {
+	// State may lag behind blocks after upgrading from a version that only
+	// deleted blocks in all mode, or after a failed pruning operation.
+	start := min(lastBlock, lastState)
+	if start >= targetHeight {
+		return nil
+	}
+	end := start + min(p.calculateBatchSize(), targetHeight-start)
+	for start < end {
+		if err := p.ctx.Err(); err != nil {
+			return err
+		}
+		batchEnd := start + min(maxPruningBatchSize, end-start)
+		if lastBlock < batchEnd {
+			if err := p.store.PruneBlocks(p.ctx, batchEnd); err != nil {
+				return fmt.Errorf("failed to prune blocks through height %d: %w", batchEnd, err)
+			}
+			lastBlock = batchEnd
+		}
+		if err := p.pruneState(lastState, batchEnd); err != nil {
 			return err
 		}
+		lastState = max(lastState, batchEnd)
+		start = batchEnd
 	}
 
-	p.logger.Debug().Uint64("pruned_up_to_height", batchEnd).Bool("da_enabled", p.daEnabled).Msg("pruned blocks up to height")
+	p.logger.Debug().Uint64("pruned_up_to_height", end).Msg("pruned blocks and state snapshots")
 	return nil
 }
 
-// calculateBatchSize returns the appropriate batch size for pruning operations.
-// The batch size is based on the pruning interval and block time to ensure reasonable progress
-// without overwhelming the node.
+// maxPruningBatchSize bounds each datastore batch, rather than the total
+// progress per interval, so fast chains can still catch up with old history.
+const maxPruningBatchSize uint64 = 10000
+
+// calculateBatchSize returns a per-interval work budget of four times the
+// expected block production. Each datastore batch is bounded independently.
 func (p *Pruner) calculateBatchSize() uint64 {
-	// Calculate batch size based on pruning interval and block time.
-	// We use 4x the blocks produced during one pruning interval as the batch size.
-	// This ensures we catch up at 3x the block production rate when there's a backlog.
-	// Example: With 100ms blocks and 15min interval: 15*60/0.1 = 9000 blocks/interval
-	//   - Batch size: 36,000 blocks (prunes 36k, chain grows 9k = net 27k catch-up per interval)
-	blocksPerInterval := uint64(p.cfg.Interval.Duration / p.blockTime)
-
-	// Ensure reasonable minimum
-	return min(blocksPerInterval*4, 10000)
+	if p.blockTime <= 0 || p.cfg.Interval.Duration <= 0 {
+		return 1
+	}
+	blocksPerInterval := max(uint64(p.cfg.Interval.Duration/p.blockTime), 1)
+	return min(blocksPerInterval, ^uint64(0)/4) * 4
 }
 
 // pruneMetadata prunes old state and execution metadata entries based on the configured retention depth.
@@ -197,37 +190,49 @@ func (p *Pruner) pruneMetadata() error {
 		return fmt.Errorf("failed to get last pruned state height: %w", err)
 	}
 
-	if lastPrunedBlock, err := p.getLastPrunedBlockHeight(p.ctx); err == nil && lastPrunedBlock > lastPrunedState {
-		lastPrunedState = lastPrunedBlock
-	}
-
 	target := height - p.cfg.KeepRecent
 	if target <= lastPrunedState {
 		return nil
 	}
 
-	batchSize := p.calculateBatchSize()
+	end := lastPrunedState + min(p.calculateBatchSize(), target-lastPrunedState)
+	for lastPrunedState < end {
+		if err := p.ctx.Err(); err != nil {
+			return err
+		}
+		batchEnd := lastPrunedState + min(maxPruningBatchSize, end-lastPrunedState)
+		if err := p.pruneState(lastPrunedState, batchEnd); err != nil {
+			return err
+		}
+		lastPrunedState = batchEnd
+	}
 
-	// prune in batches to avoid overwhelming the system
-	batchEnd := min(lastPrunedState+batchSize, target)
+	p.logger.Debug().Uint64("pruned_to", end).Msg("pruned state height metadata up to height")
+	return nil
+}
 
-	for h := lastPrunedState + 1; h <= batchEnd; h++ {
-		if err := p.store.DeleteStateAtHeight(p.ctx, h); err != nil && !errors.Is(err, ds.ErrNotFound) {
+// pruneState retries execution pruning before advancing the state cursor, so
+// failures remain recoverable even if block deletion has already committed.
+func (p *Pruner) pruneState(lastPruned, end uint64) error {
+	for h := lastPruned; h < end; {
+		h++
+		if err := p.ctx.Err(); err != nil {
 			return err
 		}
+		if err := p.store.DeleteStateAtHeight(p.ctx, h); err != nil && !errors.Is(err, ds.ErrNotFound) {
+			return fmt.Errorf("failed to prune state at height %d: %w", h, err)
+		}
 	}
-
 	if p.execPruner != nil {
-		if err := p.execPruner.PruneExec(p.ctx, batchEnd); err != nil && !errors.Is(err, ds.ErrNotFound) {
-			return err
+		if err := p.execPruner.PruneExec(p.ctx, end); err != nil && !errors.Is(err, ds.ErrNotFound) {
+			return fmt.Errorf("failed to prune execution metadata through height %d: %w", end, err)
 		}
 	}
-
-	if err := p.setLastPrunedStateHeight(p.ctx, batchEnd); err != nil {
-		return fmt.Errorf("failed to set last pruned block height: %w", err)
+	if end > lastPruned {
+		if err := p.setLastPrunedStateHeight(p.ctx, end); err != nil {
+			return fmt.Errorf("failed to set last pruned state height: %w", err)
+		}
 	}
-
-	p.logger.Debug().Uint64("pruned_to", batchEnd).Msg("pruned state height metadata up to height")
 	return nil
 }
 
@@ -239,8 +244,11 @@ func (p *Pruner) getLastPrunedBlockHeight(ctx context.Context) (uint64, error) {
 		return 0, nil
 	}
 
-	if err != nil || len(lastPrunedBlockHeightBz) != 8 {
-		return 0, fmt.Errorf("failed to get last pruned block height or invalid format: %w", err)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get last pruned block height: %w", err)
+	}
+	if len(lastPrunedBlockHeightBz) != 8 {
+		return 0, errors.New("invalid last pruned block height format")
 	}
 
 	lastPrunedBlockHeight := binary.LittleEndian.Uint64(lastPrunedBlockHeightBz)
@@ -259,13 +267,16 @@ func (p *Pruner) getLastPrunedStateHeight(ctx context.Context) (uint64, error) {
 		return 0, nil
 	}
 
-	if err != nil || len(lastPrunedStateHeightBz) != 8 {
-		return 0, fmt.Errorf("failed to get last pruned block height or invalid format: %w", err)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get last pruned state height: %w", err)
+	}
+	if len(lastPrunedStateHeightBz) != 8 {
+		return 0, errors.New("invalid last pruned state height format")
 	}
 
 	lastPrunedStateHeight := binary.LittleEndian.Uint64(lastPrunedStateHeightBz)
 	if lastPrunedStateHeight == 0 {
-		return 0, fmt.Errorf("invalid last pruned block height")
+		return 0, fmt.Errorf("invalid last pruned state height")
 	}
 
 	return lastPrunedStateHeight, nil
diff --git a/block/internal/pruner/pruner_test.go b/block/internal/pruner/pruner_test.go
index 8f0ad4773e..1fea1e5f87 100644
--- a/block/internal/pruner/pruner_test.go
+++ b/block/internal/pruner/pruner_test.go
@@ -2,6 +2,8 @@ package pruner
 
 import (
 	"context"
+	"encoding/binary"
+	"errors"
 	"testing"
 	"time"
 
@@ -51,7 +53,8 @@ func TestPrunerPruneMetadata(t *testing.T) {
 		KeepRecent: 1,
 	}
 
-	pruner := New(zerolog.New(zerolog.NewTestWriter(t)), stateStore, execAdapter, cfg, 100*time.Millisecond, "") // Empty DA address
+	pruner := New(zerolog.New(zerolog.NewTestWriter(t)), stateStore, execAdapter, cfg, 100*time.Millisecond)
+	pruner.ctx = ctx
 	require.NoError(t, pruner.pruneMetadata())
 
 	_, err := stateStore.GetStateAtHeight(ctx, 1)
@@ -90,14 +93,15 @@ func TestPrunerPruneBlocksWithoutDA(t *testing.T) {
 		execAdapter.existing[h] = struct{}{}
 	}
 
-	// Test with empty DA address (DA disabled) - should prune successfully
+	// Prune by local retention.
 	cfg := config.PruningConfig{
 		Mode:       config.PruningModeAll,
 		Interval:   config.DurationWrapper{Duration: 1 * time.Second},
 		KeepRecent: 10,
 	}
 
-	pruner := New(zerolog.New(zerolog.NewTestWriter(t)), stateStore, execAdapter, cfg, 100*time.Millisecond, "") // Empty DA address = DA disabled
+	pruner := New(zerolog.New(zerolog.NewTestWriter(t)), stateStore, execAdapter, cfg, 100*time.Millisecond)
+	pruner.ctx = ctx
 	require.NoError(t, pruner.pruneBlocks())
 
 	// Verify blocks were pruned (batch size is 40 blocks: 1s interval / 100ms block time * 4)
@@ -110,6 +114,8 @@ func TestPrunerPruneBlocksWithoutDA(t *testing.T) {
 	for h := uint64(1); h <= 40; h++ {
 		_, _, err := stateStore.GetBlockData(ctx, h)
 		require.Error(t, err, "expected block data at height %d to be pruned", h)
+		_, err = stateStore.GetStateAtHeight(ctx, h)
+		require.ErrorIs(t, err, ds.ErrNotFound)
 	}
 
 	// Verify blocks after batch were kept
@@ -125,7 +131,7 @@ func TestPrunerPruneBlocksWithoutDA(t *testing.T) {
 	}
 }
 
-func TestPrunerPruneBlocksWithDAEnabled(t *testing.T) {
+func TestPrunerPruneBlocksWithoutDAInclusionMetadata(t *testing.T) {
 	t.Parallel()
 
 	ctx := t.Context()
@@ -146,20 +152,176 @@ func TestPrunerPruneBlocksWithDAEnabled(t *testing.T) {
 		require.NoError(t, batch.Commit())
 	}
 
-	// Test with DA address provided (DA enabled) - should skip pruning when DA height is not available
+	// Missing DA inclusion metadata must not block local retention.
 	cfg := config.PruningConfig{
 		Mode:       config.PruningModeAll,
 		Interval:   config.DurationWrapper{Duration: 1 * time.Second},
 		KeepRecent: 10,
 	}
 
-	pruner := New(zerolog.New(zerolog.NewTestWriter(t)), stateStore, nil, cfg, 100*time.Millisecond, "localhost:1234") // DA enabled
-	// Should return nil (skip pruning) since DA height is not available
+	pruner := New(zerolog.New(zerolog.NewTestWriter(t)), stateStore, nil, cfg, 100*time.Millisecond)
+	pruner.ctx = ctx
+	// Process all batches until the local retention target is reached.
 	require.NoError(t, pruner.pruneBlocks())
 
-	// Verify no blocks were pruned (all blocks should still be retrievable)
+	for range 2 {
+		require.NoError(t, pruner.pruneBlocks())
+	}
 	for h := uint64(1); h <= 100; h++ {
-		_, _, err := stateStore.GetBlockData(ctx, h)
-		require.NoError(t, err, "expected block data at height %d to still exist (no pruning should have happened)", h)
+		_, _, blockErr := stateStore.GetBlockData(ctx, h)
+		_, stateErr := stateStore.GetStateAtHeight(ctx, h)
+		if h <= 90 {
+			require.ErrorIs(t, blockErr, ds.ErrNotFound)
+			require.ErrorIs(t, stateErr, ds.ErrNotFound)
+		} else {
+			require.NoError(t, blockErr)
+			require.NoError(t, stateErr)
+		}
+	}
+	state, err := stateStore.GetState(ctx)
+	require.NoError(t, err)
+	require.Equal(t, uint64(100), state.LastBlockHeight)
+}
+
+// recordingStore verifies the batch boundaries without allocating thousands of blocks.
+type recordingStore struct {
+	store.Store
+	blockBatches []uint64
+	states       []uint64
+}
+
+func (s *recordingStore) PruneBlocks(ctx context.Context, height uint64) error {
+	s.blockBatches = append(s.blockBatches, height)
+	bz := make([]byte, 8)
+	binary.LittleEndian.PutUint64(bz, height)
+	return s.SetMetadata(ctx, store.LastPrunedBlockHeightKey, bz)
+}
+
+func (s *recordingStore) DeleteStateAtHeight(_ context.Context, height uint64) error {
+	s.states = append(s.states, height)
+	return nil
+}
+
+func TestPrunerCatchupUsesBoundedBatches(t *testing.T) {
+	ctx := t.Context()
+	st := &recordingStore{Store: store.New(dssync.MutexWrap(ds.NewMapDatastore()))}
+	batch, err := st.NewBatch(ctx)
+	require.NoError(t, err)
+	require.NoError(t, batch.SetHeight(100000))
+	require.NoError(t, batch.Commit())
+	p := New(zerolog.Nop(), st, nil, config.PruningConfig{
+		Mode: config.PruningModeAll, KeepRecent: 10,
+		Interval: config.DurationWrapper{Duration: 15 * time.Minute},
+	}, 100*time.Millisecond)
+	p.ctx = ctx
+	require.NoError(t, p.pruneBlocks())
+	require.Equal(t, []uint64{10000, 20000, 30000, 36000}, st.blockBatches)
+	require.Len(t, st.states, 36000)
+
+	// The chain grows by 9000 blocks per interval; pruning must exceed that.
+	batch, err = st.NewBatch(ctx)
+	require.NoError(t, err)
+	require.NoError(t, batch.SetHeight(109000))
+	require.NoError(t, batch.Commit())
+	require.NoError(t, p.pruneBlocks())
+	require.Equal(t, []uint64{10000, 20000, 30000, 36000, 46000, 56000, 66000, 72000}, st.blockBatches)
+	require.Len(t, st.states, 72000)
+}
+
+func TestPrunerRecoversSnapshotsAfterBlockOnlyPruning(t *testing.T) {
+	for _, mode := range []string{config.PruningModeAll, config.PruningModeMetadata} {
+		t.Run(mode, func(t *testing.T) {
+			ctx := t.Context()
+			st := store.New(dssync.MutexWrap(ds.NewMapDatastore()))
+			for h := uint64(1); h <= 5; h++ {
+				batch, err := st.NewBatch(ctx)
+				require.NoError(t, err)
+				require.NoError(t, batch.SetHeight(h))
+				require.NoError(t, batch.UpdateState(types.State{LastBlockHeight: h}))
+				require.NoError(t, batch.Commit())
+			}
+			require.NoError(t, st.PruneBlocks(ctx, 4))
+			p := New(zerolog.Nop(), st, nil, config.PruningConfig{
+				Mode: mode, KeepRecent: 1,
+				Interval: config.DurationWrapper{Duration: time.Second},
+			}, 100*time.Millisecond)
+			p.ctx = ctx
+			if mode == config.PruningModeAll {
+				require.NoError(t, p.pruneBlocks())
+			} else {
+				require.NoError(t, p.pruneMetadata())
+			}
+			for h := uint64(1); h <= 4; h++ {
+				_, err := st.GetStateAtHeight(ctx, h)
+				require.ErrorIs(t, err, ds.ErrNotFound)
+			}
+			state, err := st.GetState(ctx)
+			require.NoError(t, err)
+			require.Equal(t, uint64(5), state.LastBlockHeight)
+			_, err = st.GetStateAtHeight(ctx, 5)
+			require.NoError(t, err)
+		})
 	}
 }
+
+type failingExecPruner struct {
+	err   error
+	calls []uint64
+}
+
+func (e *failingExecPruner) PruneExec(_ context.Context, height uint64) error {
+	e.calls = append(e.calls, height)
+	return e.err
+}
+
+func TestPrunerRetriesExecutionFailure(t *testing.T) {
+	ctx := t.Context()
+	st := &recordingStore{Store: store.New(dssync.MutexWrap(ds.NewMapDatastore()))}
+	batch, err := st.NewBatch(ctx)
+	require.NoError(t, err)
+	require.NoError(t, batch.SetHeight(5))
+	require.NoError(t, batch.Commit())
+	exec := &failingExecPruner{err: errors.New("execution pruning failed")}
+	cfg := config.PruningConfig{Mode: config.PruningModeAll, KeepRecent: 1,
+		Interval: config.DurationWrapper{Duration: time.Second}}
+	p := New(zerolog.Nop(), st, exec, cfg, 100*time.Millisecond)
+	p.ctx = ctx
+	require.ErrorIs(t, p.pruneBlocks(), exec.err)
+	_, err = st.GetMetadata(ctx, store.LastPrunedStateHeightKey)
+	require.ErrorIs(t, err, ds.ErrNotFound)
+
+	// A restarted pruner must retry execution despite the advanced block cursor.
+	exec.err = nil
+	p = New(zerolog.Nop(), st, exec, cfg, 100*time.Millisecond)
+	p.ctx = ctx
+	require.NoError(t, p.pruneBlocks())
+	require.Equal(t, []uint64{4, 4}, exec.calls)
+	require.Equal(t, []uint64{4}, st.blockBatches)
+}
+
+func TestPrunerPrunesBeyondStalledDAInclusion(t *testing.T) {
+	ctx := t.Context()
+	st := &recordingStore{Store: store.New(dssync.MutexWrap(ds.NewMapDatastore()))}
+	batch, err := st.NewBatch(ctx)
+	require.NoError(t, err)
+	require.NoError(t, batch.SetHeight(100))
+	require.NoError(t, batch.Commit())
+	bz := make([]byte, 8)
+	binary.LittleEndian.PutUint64(bz, 20)
+	require.NoError(t, st.SetMetadata(ctx, store.DAIncludedHeightKey, bz))
+	p := New(zerolog.Nop(), st, nil, config.PruningConfig{
+		Mode: config.PruningModeAll, KeepRecent: 10,
+		Interval: config.DurationWrapper{Duration: time.Second},
+	}, 100*time.Millisecond)
+	p.ctx = ctx
+	require.NoError(t, p.pruneBlocks())
+	require.Equal(t, []uint64{40}, st.blockBatches)
+	require.NoError(t, p.pruneBlocks())
+	require.NoError(t, p.pruneBlocks())
+	require.Equal(t, []uint64{40, 80, 90}, st.blockBatches)
+	require.Len(t, st.states, 90)
+	// Inclusion progress stays at 20; pruning does not pretend DA recovered.
+	included, err := st.GetMetadata(ctx, store.DAIncludedHeightKey)
+	require.NoError(t, err)
+	require.Equal(t, uint64(20), binary.LittleEndian.Uint64(included))
+}
diff --git a/block/internal/syncing/da_follower.go b/block/internal/syncing/da_follower.go
index 443fb876ae..bdc12ea5b5 100644
--- a/block/internal/syncing/da_follower.go
+++ b/block/internal/syncing/da_follower.go
@@ -3,6 +3,7 @@ package syncing
 import (
 	"context"
 	"errors"
+	"fmt"
 	"slices"
 	"sync"
 	"time"
@@ -30,6 +31,11 @@ type daFollower struct {
 	retriever  DARetriever
 	eventSink  common.EventSink
 	logger     zerolog.Logger
+	onSkip     func(context.Context, uint64) error
+
+	// Accessed only by the sequential catch-up goroutine.
+	failedHeight  uint64
+	fetchFailures uint8
 
 	// Priority queue for P2P hint heights (absorbed from DARetriever refactoring #2).
 	priorityMu      sync.Mutex
@@ -38,6 +44,9 @@ type daFollower struct {
 
 const maxPriorityHeights = 1024
 
+// maxDAFetchAttempts bounds retries of unavailable historical DA heights.
+const maxDAFetchAttempts uint8 = 10
+
 // DAFollowerConfig holds configuration for creating a DAFollower.
 type DAFollowerConfig struct {
 	Client        da.Client
@@ -48,6 +57,9 @@ type DAFollowerConfig struct {
 	DataNamespace []byte // may be nil or equal to Namespace
 	StartDAHeight uint64
 	DABlockTime   time.Duration
+	// OnSkip persists a height skipped after repeated retrieval failures.
+	// Returning an error keeps catch-up at that height until persistence succeeds.
+	OnSkip func(context.Context, uint64) error
 }
 
 // NewDAFollower creates a new daFollower.
@@ -58,6 +70,7 @@ func NewDAFollower(cfg DAFollowerConfig) DAFollower {
 	}
 
 	f := &daFollower{
+		onSkip:          cfg.OnSkip,
 		retriever:       cfg.Retriever,
 		eventSink:       cfg.EventSink,
 		logger:          cfg.Logger.With().Str("component", "da_follower").Logger(),
@@ -126,54 +139,88 @@ func (f *daFollower) HandleEvent(ctx context.Context, ev datypes.SubscriptionEve
 func (f *daFollower) HandleCatchup(ctx context.Context, daHeight uint64) error {
 	// 1. Drain stale or future priority heights from P2P hints
 	for priorityHeight := f.popPriorityHeight(); priorityHeight != 0; priorityHeight = f.popPriorityHeight() {
-		if priorityHeight < daHeight {
-			continue // skip stale hints without yielding back to the catchup loop
+		if priorityHeight <= daHeight {
+			continue // sequential retrieval handles the current height
 		}
 
 		f.logger.Debug().
 			Uint64("da_height", priorityHeight).
 			Msg("fetching priority DA height from P2P hint")
 
-		if err := f.fetchAndPipeHeight(ctx, priorityHeight); err != nil {
+		if retrievalFailed, err := f.fetchAndPipeHeight(ctx, priorityHeight); err != nil {
 			if errors.Is(err, datypes.ErrHeightFromFuture) {
 				// Priority hint points to a future height — silently ignore.
 				f.logger.Debug().Uint64("priority_da_height", priorityHeight).
 					Msg("priority hint is from future, ignoring")
 				continue
 			}
-			// Roll back so daHeight is attempted again next cycle after backoff.
-			return err
+			if retrievalFailed {
+				f.logger.Warn().Err(err).Uint64("priority_da_height", priorityHeight).
+					Msg("priority DA retrieval failed, continuing sequential catch-up")
+				break
+			}
+			return err // event delivery failures must still be retried
 		}
 		break // continue with daHeight
 	}
 
-	// 2. Normal sequential fetch
-	if err := f.fetchAndPipeHeight(ctx, daHeight); err != nil {
+	// 2. Normal sequential fetch. Only retrieval failures count toward the
+	// limit: event delivery must succeed, and future heights must remain pending.
+	retrievalFailed, err := f.fetchAndPipeHeight(ctx, daHeight)
+	if !retrievalFailed {
+		f.fetchFailures = 0
 		return err
 	}
-	return nil
+	if ctx.Err() != nil || errors.Is(err, datypes.ErrHeightFromFuture) {
+		return err
+	}
+	if f.failedHeight != daHeight {
+		f.failedHeight = daHeight
+		f.fetchFailures = 0
+	}
+	if f.fetchFailures < maxDAFetchAttempts {
+		f.fetchFailures++
+	}
+	if f.fetchFailures < maxDAFetchAttempts {
+		return err // subscriber backs off and retries the same height
+	}
+	// During an outage the error may not identify a future height. Never
+	// skip past the observed head just because the transport is unavailable.
+	if f.subscriber != nil && daHeight > f.subscriber.HighestSeenDAHeight() {
+		return err
+	}
+	if f.onSkip != nil {
+		if skipErr := f.onSkip(ctx, daHeight); skipErr != nil {
+			return fmt.Errorf("persist skipped DA height %d: %w", daHeight, skipErr)
+		}
+	}
+	f.logger.Warn().Err(err).Uint64("da_height", daHeight).
+		Uint8("attempts", f.fetchFailures).Msg("skipping DA height after repeated retrieval failures")
+	f.fetchFailures = 0
+	return nil // subscriber advances to the next DA height
 }
 
-// fetchAndPipeHeight retrieves events at a single DA height and pipes them.
+// fetchAndPipeHeight retrieves and pipes events, reporting whether an error
+// came from retrieval so event delivery failures cannot trigger a skipped height.
 // It does NOT handle ErrHeightFromFuture — callers must decide how to react
 // because the correct response depends on whether this is a normal sequential
 // catchup or a priority-hint fetch.
-func (f *daFollower) fetchAndPipeHeight(ctx context.Context, daHeight uint64) error {
+func (f *daFollower) fetchAndPipeHeight(ctx context.Context, daHeight uint64) (retrievalFailed bool, err error) {
 	events, err := f.retriever.RetrieveFromDA(ctx, daHeight)
 	if err != nil {
 		if errors.Is(err, datypes.ErrBlobNotFound) {
-			return nil
+			return false, nil
 		}
-		return err
+		return true, err
 	}
 
 	for _, event := range events {
 		if err := f.eventSink.PipeEvent(ctx, event); err != nil {
-			return err
+			return false, err
 		}
 	}
 
-	return nil
+	return false, nil
 }
 
 // QueuePriorityHeight queues a DA height for priority retrieval.
diff --git a/block/internal/syncing/da_follower_test.go b/block/internal/syncing/da_follower_test.go
index 710d2d81d0..f1781ce148 100644
--- a/block/internal/syncing/da_follower_test.go
+++ b/block/internal/syncing/da_follower_test.go
@@ -3,7 +3,9 @@ package syncing
 import (
 	"context"
 	"errors"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
@@ -12,6 +14,7 @@ import (
 
 	"github.com/evstack/ev-node/block/internal/common"
 	datypes "github.com/evstack/ev-node/pkg/da/types"
+	testmocks "github.com/evstack/ev-node/test/mocks"
 )
 
 func TestDAFollower_HandleEvent(t *testing.T) {
@@ -251,3 +254,148 @@ func makeRange(start, end uint64) []uint64 {
 	}
 	return out
 }
+
+func TestDAFollowerSkipsAfterTenFailedFetchesAndContinues(t *testing.T) {
+	retriever := NewMockDARetriever(t)
+	fetchErr := errors.New("historical DA height no longer available")
+	for _, h := range []uint64{100, 101} {
+		retriever.On("RetrieveFromDA", mock.Anything, h).Return(nil, fetchErr).Times(10)
+	}
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(102)).Return([]common.DAHeightEvent(nil), nil).Once()
+	client := testmocks.NewMockClient(t)
+	client.On("SupportsSubscribe").Return(false)
+	client.On("GetLatestDAHeight", mock.Anything).Return(uint64(102), nil)
+	var lastSkipped atomic.Uint64
+	f := NewDAFollower(DAFollowerConfig{
+		Client: client, Retriever: retriever, Logger: zerolog.Nop(),
+		Namespace: []byte("ns"), StartDAHeight: 100, DABlockTime: time.Millisecond,
+		OnSkip: func(_ context.Context, height uint64) error {
+			lastSkipped.Store(height)
+			return nil
+		},
+	}).(*daFollower)
+	require.NoError(t, f.Start(t.Context()))
+	t.Cleanup(f.Stop)
+	require.Eventually(t, func() bool { return f.subscriber.LocalDAHeight() == 103 }, time.Second, time.Millisecond)
+	require.Equal(t, uint64(101), lastSkipped.Load())
+	// Stop before checking mocks so no background calls race with assertions.
+	f.Stop()
+	retriever.AssertExpectations(t)
+}
+
+func TestDAFollowerDoesNotSkipFutureCancellationOrDeliveryFailures(t *testing.T) {
+	for _, kind := range []string{"future", "cancellation", "delivery"} {
+		t.Run(kind, func(t *testing.T) {
+			ctx := t.Context()
+			retriever := NewMockDARetriever(t)
+			var expected error
+			var events []common.DAHeightEvent
+			var retrievalErr error
+			switch kind {
+			case "future":
+				expected = datypes.ErrHeightFromFuture
+				retrievalErr = expected
+			case "cancellation":
+				canceled, cancel := context.WithCancel(ctx)
+				cancel()
+				ctx = canceled
+				expected = context.Canceled
+				retrievalErr = expected
+			case "delivery":
+				expected = errors.New("event delivery failed")
+				events = []common.DAHeightEvent{{DaHeight: 100}}
+			}
+			retriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return(events, retrievalErr).Times(12)
+			f := &daFollower{retriever: retriever, logger: zerolog.Nop(),
+				eventSink: common.EventSinkFunc(func(context.Context, common.DAHeightEvent) error { return expected }),
+				onSkip: func(context.Context, uint64) error {
+					t.Fatal("height must not be skipped")
+					return nil
+				},
+			}
+			for range 12 {
+				require.ErrorIs(t, f.HandleCatchup(ctx, 100), expected)
+			}
+			require.Zero(t, f.fetchFailures)
+		})
+	}
+}
+
+func TestDAFollowerSkipWaitsForPersistence(t *testing.T) {
+	retriever := NewMockDARetriever(t)
+	fetchErr := errors.New("DA unavailable")
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return(nil, fetchErr).Times(11)
+	persistErr := errors.New("metadata write failed")
+	f := &daFollower{retriever: retriever, logger: zerolog.Nop(),
+		onSkip: func(context.Context, uint64) error { return persistErr },
+	}
+	for range 9 {
+		require.ErrorIs(t, f.HandleCatchup(t.Context(), 100), fetchErr)
+	}
+	require.ErrorIs(t, f.HandleCatchup(t.Context(), 100), persistErr)
+	f.onSkip = func(context.Context, uint64) error { return nil }
+	require.NoError(t, f.HandleCatchup(t.Context(), 100))
+}
+
+func TestDAFollowerResetsFetchFailuresAfterRecovery(t *testing.T) {
+	retriever := NewMockDARetriever(t)
+	fetchErr := errors.New("DA unavailable")
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return(nil, fetchErr).Times(9)
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return([]common.DAHeightEvent(nil), nil).Once()
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(101)).Return(nil, fetchErr).Times(10)
+	var skipped []uint64
+	f := &daFollower{retriever: retriever, logger: zerolog.Nop(),
+		onSkip: func(_ context.Context, h uint64) error { skipped = append(skipped, h); return nil },
+	}
+	for range 9 {
+		require.ErrorIs(t, f.HandleCatchup(t.Context(), 100), fetchErr)
+	}
+	require.NoError(t, f.HandleCatchup(t.Context(), 100))
+	for range 9 {
+		require.ErrorIs(t, f.HandleCatchup(t.Context(), 101), fetchErr)
+	}
+	require.Empty(t, skipped)
+	require.NoError(t, f.HandleCatchup(t.Context(), 101))
+	require.Equal(t, []uint64{101}, skipped)
+}
+
+func TestDAFollowerPriorityFailuresDoNotPreventSequentialRetries(t *testing.T) {
+	retriever := NewMockDARetriever(t)
+	fetchErr := errors.New("DA unavailable")
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return(nil, fetchErr).Times(10)
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(105)).Return(nil, fetchErr).Times(10)
+	var skipped []uint64
+	f := &daFollower{retriever: retriever, logger: zerolog.Nop(),
+		onSkip: func(_ context.Context, h uint64) error { skipped = append(skipped, h); return nil },
+	}
+	for attempt := range 10 {
+		// Repeated hints for the current and future heights must not starve
+		// sequential retrieval or cause a second fetch of the current height.
+		f.QueuePriorityHeight(100)
+		f.QueuePriorityHeight(105)
+		err := f.HandleCatchup(t.Context(), 100)
+		if attempt < 9 {
+			require.ErrorIs(t, err, fetchErr)
+		} else {
+			require.NoError(t, err)
+		}
+	}
+	require.Equal(t, []uint64{100}, skipped)
+}
+
+func TestDAFollowerDoesNotSkipBeyondObservedHeadDuringOutage(t *testing.T) {
+	retriever := NewMockDARetriever(t)
+	fetchErr := errors.New("DA transport unavailable")
+	retriever.On("RetrieveFromDA", mock.Anything, uint64(101)).Return(nil, fetchErr).Times(12)
+	f := NewDAFollower(DAFollowerConfig{
+		Retriever: retriever, Logger: zerolog.Nop(), Namespace: []byte("ns"), StartDAHeight: 100,
+		OnSkip: func(context.Context, uint64) error {
+			t.Fatal("must not skip beyond the observed DA head")
+			return nil
+		},
+	}).(*daFollower)
+	for range 12 {
+		require.ErrorIs(t, f.HandleCatchup(t.Context(), 101), fetchErr)
+	}
+	require.Equal(t, maxDAFetchAttempts, f.fetchFailures)
+}
diff --git a/block/internal/syncing/da_retriever.go b/block/internal/syncing/da_retriever.go
index 7f395d041f..a645a6de1c 100644
--- a/block/internal/syncing/da_retriever.go
+++ b/block/internal/syncing/da_retriever.go
@@ -161,8 +161,12 @@ func (r *daRetriever) validateBlobResponse(res datypes.ResultRetrieve, daHeight
 	case datypes.StatusSuccess:
 		r.logger.Debug().Uint64("da_height", daHeight).Msg("successfully retrieved from DA")
 		return nil
+	case datypes.StatusContextDeadline:
+		return fmt.Errorf("DA retrieval failed: %w: %s", context.DeadlineExceeded, res.Message)
+	case datypes.StatusContextCanceled:
+		return fmt.Errorf("DA retrieval failed: %w: %s", context.Canceled, res.Message)
 	default:
-		return nil
+		return fmt.Errorf("DA retrieval failed: unexpected status %d: %s", res.Code, res.Message)
 	}
 }
 
diff --git a/block/internal/syncing/da_retriever_test.go b/block/internal/syncing/da_retriever_test.go
index 2feb4577aa..a38c1d67b7 100644
--- a/block/internal/syncing/da_retriever_test.go
+++ b/block/internal/syncing/da_retriever_test.go
@@ -469,3 +469,34 @@ func Test_isEmptyDataExpected(t *testing.T) {
 	h.DataHash = common.DataHashForEmptyTxs
 	assert.True(t, isEmptyDataExpected(h))
 }
+
+func TestDARetrieverRejectsFailedReadsInEitherNamespace(t *testing.T) {
+	for _, code := range []datypes.StatusCode{datypes.StatusUnknown, datypes.StatusContextDeadline, datypes.StatusContextCanceled} {
+		for _, failedNS := range []string{"test-ns", "test-data-ns"} {
+			t.Run(fmt.Sprintf("%d/%s", code, failedNS), func(t *testing.T) {
+				client := mocks.NewMockClient(t)
+				for _, ns := range []string{"test-ns", "test-data-ns"} {
+					result := datypes.ResultRetrieve{BaseResult: datypes.BaseResult{Code: datypes.StatusNotFound}}
+					if ns == failedNS {
+						result.Code = code
+					}
+					client.On("RetrieveBlobs", mock.Anything, uint64(42), []byte(ns)).Return(result).Once()
+				}
+				cfg := config.DefaultConfig()
+				cfg.DA.Namespace = "test-ns"
+				cfg.DA.DataNamespace = "test-data-ns"
+				r := newTestDARetriever(t, client, cfg, genesis.Genesis{})
+				events, err := r.RetrieveFromDA(t.Context(), 42)
+				require.Error(t, err)
+				require.NotErrorIs(t, err, datypes.ErrBlobNotFound)
+				require.Empty(t, events)
+				switch code {
+				case datypes.StatusContextDeadline:
+					require.ErrorIs(t, err, context.DeadlineExceeded)
+				case datypes.StatusContextCanceled:
+					require.ErrorIs(t, err, context.Canceled)
+				}
+			})
+		}
+	}
+}
diff --git a/block/internal/syncing/syncer.go b/block/internal/syncing/syncer.go
index aba3d8e44a..76b5fa05dd 100644
--- a/block/internal/syncing/syncer.go
+++ b/block/internal/syncing/syncer.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/sha256"
+	"encoding/binary"
 	"encoding/hex"
 	"errors"
 	"fmt"
@@ -14,6 +15,7 @@ import (
 	"time"
 
 	"github.com/celestiaorg/go-header"
+	ds "github.com/ipfs/go-datastore"
 	"github.com/rs/zerolog"
 
 	"github.com/evstack/ev-node/block/internal/cache"
@@ -225,6 +227,7 @@ func (s *Syncer) Start(ctx context.Context) (err error) {
 			Namespace:     s.daClient.GetHeaderNamespace(),
 			DataNamespace: s.daClient.GetDataNamespace(),
 			StartDAHeight: s.daRetrieverHeight.Load(),
+			OnSkip:        s.recordSkippedDAHeight,
 			DABlockTime:   s.config.DA.BlockTime.Duration,
 		})
 		if err = s.daFollower.Start(ctx); err != nil {
@@ -317,6 +320,21 @@ func (s *Syncer) expectedProposerForHeight(height uint64) ([]byte, bool) {
 	return state.NextProposerAddress, true
 }
 
+// recordSkippedDAHeight saves skip progress synchronously before the follower
+// advances, without changing executed state or DA inclusion progress.
+func (s *Syncer) recordSkippedDAHeight(ctx context.Context, height uint64) error {
+	if height == math.MaxUint64 {
+		return errors.New("cannot skip maximum DA height")
+	}
+	bz := make([]byte, 8)
+	binary.LittleEndian.PutUint64(bz, height)
+	if err := s.store.BatchMetadata(ctx, []store.MetadataKV{{Key: store.LastSkippedDAHeightKey, Value: bz}}, nil); err != nil {
+		return err
+	}
+	s.daRetrieverHeight.Store(max(s.daRetrieverHeight.Load(), height+1))
+	return nil
+}
+
 // initializeState loads the current sync state
 func (s *Syncer) initializeState() error {
 	// Load state from store
@@ -396,6 +414,19 @@ func (s *Syncer) initializeState() error {
 		daHeight = max(daHeight, s.cache.DaHeight())
 	}
 
+	if skipped, err := s.store.GetMetadata(s.ctx, store.LastSkippedDAHeightKey); err == nil {
+		if len(skipped) != 8 {
+			return errors.New("invalid last skipped DA height metadata")
+		}
+		lastSkipped := binary.LittleEndian.Uint64(skipped)
+		if lastSkipped == math.MaxUint64 {
+			return errors.New("last skipped DA height cannot advance beyond maximum height")
+		}
+		daHeight = max(daHeight, lastSkipped+1)
+	} else if !errors.Is(err, ds.ErrNotFound) {
+		return fmt.Errorf("load last skipped DA height: %w", err)
+	}
+
 	// dev mode for da start height
 	if startHeight := s.config.DA.StartHeight; startHeight > 0 {
 		s.logger.Info().
diff --git a/block/internal/syncing/syncer_test.go b/block/internal/syncing/syncer_test.go
index e78a8771c6..b39bcb141c 100644
--- a/block/internal/syncing/syncer_test.go
+++ b/block/internal/syncing/syncer_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	crand "crypto/rand"
 	"crypto/sha512"
+	"encoding/binary"
 	"errors"
 	"math"
 	"sync"
@@ -1457,6 +1458,8 @@ func TestSyncer_InitializeState_CallsReplayer(t *testing.T) {
 		nil,
 	)
 
+	mockStore.EXPECT().GetMetadata(mock.Anything, store.LastSkippedDAHeightKey).Return(nil, datastore.ErrNotFound)
+
 	// Mock Batch operations
 	mockBatch := testmocks.NewMockBatch(t)
 	mockBatch.Test(t)
@@ -2116,3 +2119,66 @@ func TestSyncer_Stop_DrainWorksWithoutCriticalError(t *testing.T) {
 		mockExec.AssertExpectations(t)
 	})
 }
+
+func TestSyncerRestartsPastSkippedDAHeight(t *testing.T) {
+	ctx := t.Context()
+	st := store.New(dssync.MutexWrap(datastore.NewMapDatastore()))
+	batch, err := st.NewBatch(ctx)
+	require.NoError(t, err)
+	state := types.State{ChainID: "test-chain", InitialHeight: 1, LastBlockHeight: 100, DAHeight: 20}
+	require.NoError(t, batch.SetHeight(100))
+	require.NoError(t, batch.UpdateState(state))
+	require.NoError(t, batch.Commit())
+	included := make([]byte, 8)
+	binary.LittleEndian.PutUint64(included, 20)
+	require.NoError(t, st.SetMetadata(ctx, store.DAIncludedHeightKey, included))
+
+	// Skip writes must be durable even through the asynchronous cache wrapper.
+	cached, err := store.NewCachedStore(st)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, cached.Close()) })
+	cfg := config.DefaultConfig()
+	gen := genesis.Genesis{ChainID: state.ChainID, InitialHeight: 1, DAStartHeight: 1}
+	exec := testmocks.NewMockHeightAwareExecutor(t)
+	exec.On("GetLatestHeight", mock.Anything).Return(uint64(100), nil)
+	newSyncer := func() *Syncer {
+		s := NewSyncer(cached, exec, nil, nil, common.NopMetrics(), cfg, gen, nil, nil,
+			zerolog.Nop(), common.DefaultBlockOptions(), nil, nil)
+		s.ctx = ctx
+		return s
+	}
+	s := newSyncer()
+	require.NoError(t, s.initializeState())
+	require.NoError(t, s.recordSkippedDAHeight(ctx, 50))
+	require.Equal(t, uint64(51), s.daRetrieverHeight.Load())
+	skipped, err := st.GetMetadata(ctx, store.LastSkippedDAHeightKey)
+	require.NoError(t, err)
+	require.Equal(t, uint64(50), binary.LittleEndian.Uint64(skipped))
+
+	restarted := newSyncer()
+	require.NoError(t, restarted.initializeState())
+	require.Equal(t, uint64(51), restarted.daRetrieverHeight.Load())
+	// Skip progress must not change the executed state's DA height or inclusion.
+	persisted, err := st.GetState(ctx)
+	require.NoError(t, err)
+	require.Equal(t, uint64(20), persisted.DAHeight)
+	persistedInclusion, err := st.GetMetadata(ctx, store.DAIncludedHeightKey)
+	require.NoError(t, err)
+	require.Equal(t, included, persistedInclusion)
+
+	// An explicit operator override can still select an earlier fetch height.
+	cfg.DA.StartHeight = 30
+	overridden := newSyncer()
+	require.NoError(t, overridden.initializeState())
+	require.Equal(t, uint64(30), overridden.daRetrieverHeight.Load())
+}
+
+func TestSyncerSkipPersistenceFailureDoesNotAdvance(t *testing.T) {
+	st := testmocks.NewMockStore(t)
+	writeErr := errors.New("metadata commit failed")
+	st.EXPECT().BatchMetadata(mock.Anything, mock.Anything, []string(nil)).Return(writeErr).Once()
+	s := &Syncer{store: st, daRetrieverHeight: &atomic.Uint64{}}
+	s.daRetrieverHeight.Store(100)
+	require.ErrorIs(t, s.recordSkippedDAHeight(t.Context(), 100), writeErr)
+	require.Equal(t, uint64(100), s.daRetrieverHeight.Load())
+}
diff --git a/pkg/config/config.go b/pkg/config/config.go
index f9b5ef1783..04fc15c4b9 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -382,7 +382,7 @@ const (
 
 // PruningConfig contains all pruning configuration parameters
 type PruningConfig struct {
-	Mode       string          `mapstructure:"pruning_mode" yaml:"pruning_mode" comment:"Pruning mode for stored block data and block metadata. Options: 'all' (prune all but recent blocks and their metatadas), 'metadata' (prune all but recent blocks metadatas), 'disabled' (keep all blocks and blocks metadata). Default: 'disabled'."`
+	Mode       string          `mapstructure:"pruning_mode" yaml:"pruning_mode" comment:"Pruning mode for stored block data and block metadata. Options: 'all' (prune blocks and state snapshots beyond local retention, regardless of DA inclusion), 'metadata' (prune all but recent blocks metadatas), 'disabled' (keep all blocks and blocks metadata). Default: 'disabled'."`
 	KeepRecent uint64          `mapstructure:"pruning_keep_recent" yaml:"pruning_keep_recent" comment:"Number of most recent blocks/blocks metadata to retain when pruning is enabled. Must be > 0."`
 	Interval   DurationWrapper `mapstructure:"pruning_interval" yaml:"pruning_interval" comment:"Run pruning every N minutes. Examples: \"5m\", \"10m\", \"24h\"."`
 }
diff --git a/pkg/store/cached_store.go b/pkg/store/cached_store.go
index 39031ccde3..a1da2a0c88 100644
--- a/pkg/store/cached_store.go
+++ b/pkg/store/cached_store.go
@@ -208,9 +208,17 @@ func (cs *CachedStore) GetBlockData(ctx context.Context, height uint64) (*types.
 
 // InvalidateRange removes headers in the given range from the cache.
 func (cs *CachedStore) InvalidateRange(fromHeight, toHeight uint64) {
-	for h := fromHeight; h <= toHeight; h++ {
-		cs.headerCache.Remove(h)
-		cs.blockDataCache.Remove(h)
+	// Scan the bounded caches rather than the historical height range. Pruning
+	// repeatedly invalidates from height 1, which can be far behind the head.
+	for _, h := range cs.headerCache.Keys() {
+		if h >= fromHeight && h <= toHeight {
+			cs.headerCache.Remove(h)
+		}
+	}
+	for _, h := range cs.blockDataCache.Keys() {
+		if h >= fromHeight && h <= toHeight {
+			cs.blockDataCache.Remove(h)
+		}
 	}
 }
 
diff --git a/pkg/store/cached_store_test.go b/pkg/store/cached_store_test.go
index 432d90f035..c4650a9c0e 100644
--- a/pkg/store/cached_store_test.go
+++ b/pkg/store/cached_store_test.go
@@ -400,3 +400,18 @@ func TestCachedStore_CoalescesSameKeyOps(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []byte("v2"), v, "last write (Set) should win over delete")
 }
+
+func TestCachedStoreInvalidateLargeRange(t *testing.T) {
+	cs, err := NewCachedStore(New(ds.NewMapDatastore()), WithHeaderCacheSize(3), WithBlockDataCacheSize(3))
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, cs.Close()) })
+	// A sparse cache and a huge pruning range must not require walking every
+	// height, or overflow when invalidating through the maximum height.
+	for _, h := range []uint64{1, 1 << 60, ^uint64(0)} {
+		cs.headerCache.Add(h, &types.SignedHeader{})
+		cs.blockDataCache.Add(h, &blockDataEntry{})
+	}
+	cs.InvalidateRange(2, ^uint64(0))
+	require.Equal(t, []uint64{1}, cs.headerCache.Keys())
+	require.Equal(t, []uint64{1}, cs.blockDataCache.Keys())
+}
diff --git a/pkg/store/keys.go b/pkg/store/keys.go
index 2f8d2c733b..c2170cd3ad 100644
--- a/pkg/store/keys.go
+++ b/pkg/store/keys.go
@@ -19,6 +19,10 @@ const (
 	// DAIncludedHeightKey is the key used for persisting the da included height in store.
 	DAIncludedHeightKey = "d"
 
+	// LastSkippedDAHeightKey records the last DA fetch height abandoned after
+	// repeated failures. It is independent of DA inclusion and executed state.
+	LastSkippedDAHeightKey = "last-skipped-da-height"
+
 	// LastSubmittedHeaderHeightKey is the key used for persisting the last submitted header height in store.
 	LastSubmittedHeaderHeightKey = "last-submitted-header-height"
 
diff --git a/pkg/store/store.go b/pkg/store/store.go
index 4c045d1d89..a04fea99ff 100644
--- a/pkg/store/store.go
+++ b/pkg/store/store.go
@@ -348,7 +348,7 @@ func (s *DefaultStore) PruneBlocks(ctx context.Context, height uint64) error {
 		if !errors.Is(err, ds.ErrNotFound) {
 			return fmt.Errorf("failed to get last pruned height: %w", err)
 		}
-	} else if len(meta) == heightLength {
+	} else {
 		lastPruned, err = decodeHeight(meta)
 		if err != nil {
 			return fmt.Errorf("failed to decode last pruned height: %w", err)
@@ -361,14 +361,15 @@ func (s *DefaultStore) PruneBlocks(ctx context.Context, height uint64) error {
 	}
 
 	// Delete block data for heights in (lastPruned, height].
-	for h := lastPruned + 1; h <= height; h++ {
-		// Get header blob to compute the hash index key. If header is already
-		// missing (e.g. due to previous partial pruning), just skip this height.
+	for h := lastPruned; h < height; {
+		h++
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		// Get the header for the hash index, but delete the remaining data
+		// even if the header is already missing.
 		headerBlob, err := s.db.Get(ctx, ds.NewKey(getHeaderKey(h)))
-		if err != nil {
-			if errors.Is(err, ds.ErrNotFound) {
-				continue
-			}
+		if err != nil && !errors.Is(err, ds.ErrNotFound) {
 			return fmt.Errorf("failed to get header at height %d during pruning: %w", h, err)
 		}
 
@@ -402,9 +403,9 @@ func (s *DefaultStore) PruneBlocks(ctx context.Context, height uint64) error {
 			}
 		}
 
-		headerHash := sha256.Sum256(headerBlob)
-		if err := batch.Delete(ctx, ds.NewKey(getIndexKey(headerHash[:]))); err != nil {
-			if !errors.Is(err, ds.ErrNotFound) {
+		if headerBlob != nil {
+			headerHash := sha256.Sum256(headerBlob)
+			if err := batch.Delete(ctx, ds.NewKey(getIndexKey(headerHash[:]))); err != nil && !errors.Is(err, ds.ErrNotFound) {
 				return fmt.Errorf("failed to delete index for height %d during pruning: %w", h, err)
 			}
 		}
diff --git a/pkg/store/store_test.go b/pkg/store/store_test.go
index 68903e1c7c..c278e47fdd 100644
--- a/pkg/store/store_test.go
+++ b/pkg/store/store_test.go
@@ -1292,3 +1292,33 @@ func TestRollbackDAIncludedHeightGetMetadataError(t *testing.T) {
 	require.Contains(err.Error(), "failed to get DA included height")
 	require.Contains(err.Error(), "metadata retrieval failed")
 }
+
+func TestPruneBlocksMissingHeaderStillDeletesRemainingData(t *testing.T) {
+	ctx := t.Context()
+	kv := ds.NewMapDatastore()
+	s := New(kv)
+	header := &types.SignedHeader{Header: types.Header{BaseHeader: types.BaseHeader{Height: 1}}}
+	data := &types.Data{}
+	sig := types.Signature{1}
+	batch, err := s.NewBatch(ctx)
+	require.NoError(t, err)
+	require.NoError(t, batch.SaveBlockData(header, data, &sig))
+	require.NoError(t, batch.Commit())
+	require.NoError(t, s.SetMetadata(ctx, GetHeightToDAHeightHeaderKey(1), encodeHeight(10)))
+	require.NoError(t, s.SetMetadata(ctx, GetHeightToDAHeightDataKey(1), encodeHeight(10)))
+	require.NoError(t, kv.Delete(ctx, ds.NewKey(getHeaderKey(1))))
+
+	require.NoError(t, s.PruneBlocks(ctx, 1))
+	for _, key := range []string{getDataKey(1), getSignatureKey(1),
+		GetMetaKey(GetHeightToDAHeightHeaderKey(1)), GetMetaKey(GetHeightToDAHeightDataKey(1))} {
+		_, err := kv.Get(ctx, ds.NewKey(key))
+		require.ErrorIs(t, err, ds.ErrNotFound, key)
+	}
+}
+
+func TestPruneBlocksRejectsMalformedCursor(t *testing.T) {
+	ctx := t.Context()
+	s := New(ds.NewMapDatastore())
+	require.NoError(t, s.SetMetadata(ctx, LastPrunedBlockHeightKey, []byte{1}))
+	require.Error(t, s.PruneBlocks(ctx, 5))
+}