fixes

Author: julienrbrt

block/internal/cache/pending_base.go

@@ -103,9 +103,10 @@ func (pb *pendingBase[T]) setLastSubmittedHeight(ctx context.Context, newLastSub
 		if err != nil {
 			pb.logger.Error().Err(err).Msg("failed to store height of latest item submitted to DA")
 		}
-
-		// Clear marshalled cache for submitted heights
-		pb.clearMarshalledCacheUpTo(newLastSubmittedHeight)
+		// Note: We don't explicitly clear the cache here. Instead, we use lazy invalidation
+		// in getMarshalledForHeight by checking against lastHeight. This avoids O(N)
+		// iteration over the cache on every submission. The LRU will naturally evict
+		// old entries when capacity is reached.
 	}
 }
 
@@ -128,8 +129,14 @@ func (pb *pendingBase[T]) init() error {
 	return nil
 }
 
-// getMarshalledForHeight returns cached marshalled bytes for a height, or nil if not cached
+// getMarshalledForHeight returns cached marshalled bytes for a height, or nil if not cached.
+// Uses lazy invalidation: entries at or below lastHeight are considered invalid and ignored.
 func (pb *pendingBase[T]) getMarshalledForHeight(height uint64) []byte {
+	// Lazy invalidation: don't return cached data for already-submitted heights
+	if height <= pb.lastHeight.Load() {
+		return nil
+	}
+
 	pb.marshalledCacheMu.RLock()
 	defer pb.marshalledCacheMu.RUnlock()
 
@@ -139,25 +146,16 @@ func (pb *pendingBase[T]) getMarshalledForHeight(height uint64) []byte {
 	return nil
 }
 
-// setMarshalledForHeight caches marshalled bytes for a height
+// setMarshalledForHeight caches marshalled bytes for a height.
+// Does not cache heights that have already been submitted.
 func (pb *pendingBase[T]) setMarshalledForHeight(height uint64, marshalled []byte) {
-	pb.marshalledCacheMu.Lock()
-	defer pb.marshalledCacheMu.Unlock()
-
-	pb.marshalledCache.Add(height, marshalled)
-}
+	// Don't cache already-submitted heights
+	if height <= pb.lastHeight.Load() {
+		return
+	}
 
-// clearMarshalledCacheUpTo removes cached marshalled bytes up to and including the given height.
-// With LRU cache, we iterate through keys and remove those <= height.
-func (pb *pendingBase[T]) clearMarshalledCacheUpTo(height uint64) {
 	pb.marshalledCacheMu.Lock()
 	defer pb.marshalledCacheMu.Unlock()
 
-	// Get all keys and remove those that are <= height
-	keys := pb.marshalledCache.Keys()
-	for _, h := range keys {
-		if h <= height {
-			pb.marshalledCache.Remove(h)
-		}
-	}
+	pb.marshalledCache.Add(height, marshalled)
 }

block/internal/submitting/da_submitter.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	lru "github.com/hashicorp/golang-lru/v2"
@@ -118,6 +119,10 @@ type DASubmitter struct {
 	envelopeCache   *lru.Cache[uint64, []byte]
 	envelopeCacheMu sync.RWMutex
 
+	// lastSubmittedHeight tracks the last successfully submitted height for lazy cache invalidation.
+	// This avoids O(N) iteration over the cache on every submission.
+	lastSubmittedHeight atomic.Uint64
+
 	// signingWorkers is the number of parallel workers for signing
 	signingWorkers int
 }
@@ -223,8 +228,8 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.Signed
 			if l := len(submitted); l > 0 {
 				lastHeight := submitted[l-1].Height()
 				cache.SetLastSubmittedHeaderHeight(ctx, lastHeight)
-				// Clear envelope cache for successfully submitted heights
-				s.clearEnvelopeCacheUpTo(lastHeight)
+				// Update last submitted height for lazy cache invalidation (O(1) instead of O(N))
+				s.lastSubmittedHeight.Store(lastHeight)
 			}
 		},
 		"header",
@@ -361,10 +366,15 @@ func (s *DASubmitter) signAndCacheEnvelope(header *types.SignedHeader, marshalle
 }
 
 // getCachedEnvelope retrieves a cached envelope for the given height.
+// Uses lazy invalidation: entries at or below lastSubmittedHeight are considered invalid.
 func (s *DASubmitter) getCachedEnvelope(height uint64) []byte {
 	if s.envelopeCache == nil {
 		return nil
 	}
+	// Lazy invalidation: don't return cached data for already-submitted heights
+	if height <= s.lastSubmittedHeight.Load() {
+		return nil
+	}
 	s.envelopeCacheMu.RLock()
 	defer s.envelopeCacheMu.RUnlock()
 
@@ -375,30 +385,19 @@ func (s *DASubmitter) getCachedEnvelope(height uint64) []byte {
 }
 
 // setCachedEnvelope stores an envelope in the cache.
+// Does not cache heights that have already been submitted.
 func (s *DASubmitter) setCachedEnvelope(height uint64, envelope []byte) {
 	if s.envelopeCache == nil {
 		return
 	}
-	s.envelopeCacheMu.Lock()
-	defer s.envelopeCacheMu.Unlock()
-
-	s.envelopeCache.Add(height, envelope)
-}
-
-// clearEnvelopeCacheUpTo removes cached envelopes up to and including the given height.
-func (s *DASubmitter) clearEnvelopeCacheUpTo(height uint64) {
-	if s.envelopeCache == nil {
+	// Don't cache already-submitted heights
+	if height <= s.lastSubmittedHeight.Load() {
 		return
 	}
 	s.envelopeCacheMu.Lock()
 	defer s.envelopeCacheMu.Unlock()
 
-	keys := s.envelopeCache.Keys()
-	for _, h := range keys {
-		if h <= height {
-			s.envelopeCache.Remove(h)
-		}
-	}
+	s.envelopeCache.Add(height, envelope)
 }
 
 // SubmitData submits pending data to DA layer