op-node: receipts cache limit by volume

op-node/sources/caching/cache.go

@@ -9,9 +9,10 @@ type Metrics interface {
 
 // LRUCache wraps hashicorp *lru.Cache and tracks cache metrics
 type LRUCache struct {
-	m     Metrics
-	label string
-	inner *lru.Cache
+	m           Metrics
+	label       string
+	inner       *lru.Cache
+	volumeLimit *VolumeLimit
 }
 
 func (c *LRUCache) Get(key any) (value any, ok bool) {
@@ -27,7 +28,22 @@ func (c *LRUCache) Add(key, value any) (evicted bool) {
 	if c.m != nil {
 		c.m.CacheAdd(c.label, c.inner.Len(), evicted)
 	}
-	return evicted
+
+	EvictedByVolumeLimit := false
+	if c.volumeLimit != nil {
+		c.volumeLimit.Add(key, value)
+
+		// if the cache is over the volume limit, evict the oldest items
+		for c.volumeLimit.OverLimit() {
+			_, _, ok := c.inner.RemoveOldest()
+			if !ok {
+				break
+			}
+
+			EvictedByVolumeLimit = EvictedByVolumeLimit || ok
+		}
+	}
+	return evicted || EvictedByVolumeLimit
 }
 
 // NewLRUCache creates a LRU cache with the given metrics, labeling the cache adds/gets.
@@ -36,8 +52,23 @@ func NewLRUCache(m Metrics, label string, maxSize int) *LRUCache {
 	// no errors if the size is positive
 	cache, _ := lru.New(maxSize)
 	return &LRUCache{
-		m:     m,
-		label: label,
-		inner: cache,
+		m:           m,
+		label:       label,
+		inner:       cache,
+		volumeLimit: nil,
+	}
+}
+
+func NewLRUCacheWithVolumeLimit(m Metrics, label string, maxSize int, volumeLimit VolumeLimit) *LRUCache {
+	onEvicted := func(key interface{}, value interface{}) {
+		// remove the volume of the evicted item
+		volumeLimit.Remove(key, value)
+	}
+	cache, _ := lru.NewWithEvict(maxSize, onEvicted)
+	return &LRUCache{
+		m:           m,
+		label:       label,
+		inner:       cache,
+		volumeLimit: &volumeLimit,
 	}
 }

op-node/sources/caching/volume_limit.go

@@ -0,0 +1,30 @@
+package caching
+
+type VolumeFn func(key interface{}, value interface{}) int
+
+// VolumeLimit is used to limit the total volume of the cache.
+type VolumeLimit struct {
+	volume   int
+	limit    int
+	volumeFn VolumeFn
+}
+
+func NewVolumeLimit(limit int, volumeFn VolumeFn) VolumeLimit {
+	return VolumeLimit{
+		volume:   0,
+		limit:    limit,
+		volumeFn: volumeFn,
+	}
+}
+
+func (v *VolumeLimit) Add(key interface{}, value interface{}) {
+	v.volume += v.volumeFn(key, value)
+}
+
+func (v *VolumeLimit) Remove(key interface{}, value interface{}) {
+	v.volume -= v.volumeFn(key, value)
+}
+
+func (v *VolumeLimit) OverLimit() bool {
+	return v.volume > v.limit
+}

op-node/sources/eth_client.go

@@ -165,14 +165,25 @@ func NewEthClient(client client.RPC, log log.Logger, metrics caching.Metrics, co
 		return nil, fmt.Errorf("bad config, cannot create L1 source: %w", err)
 	}
 	client = LimitRPC(client, config.MaxConcurrentRequests)
+
+	receiptsVolumeLimit := func(maxSize int) caching.VolumeLimit {
+		var EstimateReceiptsSizePerBlock = 1024 * 1024 // 1MB per block
+		return caching.NewVolumeLimit(EstimateReceiptsSizePerBlock*maxSize, func(_ interface{}, value interface{}) (volume int) {
+			for _, rec := range value.(types.Receipts) {
+				volume += int(rec.Size())
+			}
+			return volume
+		})
+	}
+
 	return &EthClient{
 		client:                  client,
 		maxBatchSize:            config.MaxRequestsPerBatch,
 		trustRPC:                config.TrustRPC,
 		mustBePostMerge:         config.MustBePostMerge,
 		provKind:                config.RPCProviderKind,
 		log:                     log,
-		receiptsCache:           caching.NewLRUCache(metrics, "receipts", config.ReceiptsCacheSize),
+		receiptsCache:           caching.NewLRUCacheWithVolumeLimit(metrics, "receipts", config.ReceiptsCacheSize, receiptsVolumeLimit(config.ReceiptsCacheSize)),
 		transactionsCache:       caching.NewLRUCache(metrics, "txs", config.TransactionsCacheSize),
 		headersCache:            caching.NewLRUCache(metrics, "headers", config.HeadersCacheSize),
 		payloadsCache:           caching.NewLRUCache(metrics, "payloads", config.PayloadsCacheSize),