feat: block time average

pkg/transaction/wrapped/cache/cache.go

@@ -13,7 +13,10 @@ import (
 )
 
 type (
-	Loader[T any]         func() (T, error)
+	// Loader loads a fresh value when the cache is stale. Prev is the last
+	// successfully cached value, read immediately before the load runs.
+	Loader[T any] func(prev T) (T, error)
+	// ReuseEvaluator reports whether a cached value may still be used.
 	ReuseEvaluator[T any] func(value T) bool
 )
 
@@ -50,6 +53,8 @@ func (c *SingleFlightCache[T]) Set(value T) {
 	c.value = value
 }
 
+// PeekOrLoad returns the cached value when canReuse allows it. Otherwise it
+// loads a new value via loader, passing the current cache entry as prev.
 func (c *SingleFlightCache[T]) PeekOrLoad(ctx context.Context, canReuse ReuseEvaluator[T], loader Loader[T]) (T, error) {
 	c.mu.RLock()
 	value := c.value
@@ -64,7 +69,12 @@ func (c *SingleFlightCache[T]) PeekOrLoad(ctx context.Context, canReuse ReuseEva
 
 	result, shared, err := c.group.Do(ctx, c.key, func(ctx context.Context) (any, error) {
 		c.metrics.Loads.Inc()
-		value, err := loader()
+
+		c.mu.RLock()
+		prev := c.value
+		c.mu.RUnlock()
+
+		value, err := loader(prev)
 		if err != nil {
 			c.metrics.LoadErrors.Inc()
 			return value, err

pkg/transaction/wrapped/cache/cache_test.go

@@ -37,7 +37,7 @@ func TestPeekOrLoadHit(t *testing.T) {
 		func(value uint64) bool {
 			return true
 		},
-		func() (uint64, error) {
+		func(_ uint64) (uint64, error) {
 			loadCount.Add(1)
 			return 0, errors.New("loader must not run")
 		},
@@ -59,7 +59,7 @@ func TestPeekOrLoadMiss(t *testing.T) {
 		func(value uint64) bool {
 			return false
 		},
-		func() (uint64, error) {
+		func(_ uint64) (uint64, error) {
 			loadCount.Add(1)
 			return 99, nil
 		},
@@ -75,7 +75,7 @@ func TestPeekOrLoadMiss(t *testing.T) {
 		func(value uint64) bool {
 			return true
 		},
-		func() (uint64, error) {
+		func(_ uint64) (uint64, error) {
 			verifyLoads.Add(1)
 			return 0, errors.New("unexpected load on verify")
 		},
@@ -97,7 +97,7 @@ func TestPeekOrLoadError(t *testing.T) {
 		func(value uint64) bool {
 			return false
 		},
-		func() (uint64, error) {
+		func(_ uint64) (uint64, error) {
 			loadCount.Add(1)
 			return 99, errLoad
 		},
@@ -112,7 +112,7 @@ func TestPeekOrLoadError(t *testing.T) {
 		func(value uint64) bool {
 			return false
 		},
-		func() (uint64, error) {
+		func(_ uint64) (uint64, error) {
 			loadCount.Add(1)
 			return 0, errLoad
 		},
@@ -143,7 +143,7 @@ func TestPeekOrLoadSingleflight(t *testing.T) {
 					func(value uint64) bool {
 						return false
 					},
-					func() (uint64, error) {
+					func(_ uint64) (uint64, error) {
 						loadCount.Add(1)
 						<-gate
 						return value, nil
@@ -186,7 +186,7 @@ func TestPeekOrLoadContextCancellation(t *testing.T) {
 				func(value uint64) bool {
 					return false
 				},
-				func() (uint64, error) {
+				func(_ uint64) (uint64, error) {
 					<-gate
 					return expectedVal, nil
 				},
@@ -200,7 +200,7 @@ func TestPeekOrLoadContextCancellation(t *testing.T) {
 				func(value uint64) bool {
 					return false
 				},
-				func() (uint64, error) {
+				func(_ uint64) (uint64, error) {
 					<-gate
 					return expectedVal, nil
 				},

pkg/transaction/wrapped/metrics.go

@@ -27,6 +27,7 @@ type metrics struct {
 	SendTransactionCalls          prometheus.Counter
 	FilterLogsCalls               prometheus.Counter
 	ChainIDCalls                  prometheus.Counter
+	AverageBlockTimeSeconds       prometheus.Gauge
 }
 
 func newMetrics() metrics {
@@ -129,6 +130,12 @@ func newMetrics() metrics {
 			Name:      "calls_chain_id",
 			Help:      "Count of eth_chainId rpc calls",
 		}),
+		AverageBlockTimeSeconds: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "average_block_time_seconds",
+			Help:      "Observed average block time in seconds, updated on each block number cache refresh.",
+		}),
 	}
 }
 

pkg/transaction/wrapped/wrapped.go

@@ -18,11 +18,14 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/transaction/wrapped/cache"
 )
 
+const maxAverageBlockTime = 30 * time.Second
+
 var _ transaction.Backend = (*wrappedBackend)(nil)
 
 type blockNumberAnchor struct {
-	number    uint64
-	timestamp time.Time
+	number           uint64
+	timestamp        time.Time
+	averageBlockTime time.Duration
 }
 
 type wrappedBackend struct {
@@ -86,7 +89,7 @@ func (b *wrappedBackend) BlockNumber(ctx context.Context) (uint64, error) {
 		return elapsedBlocks < b.blockSyncInterval
 	}
 
-	loadFreshBlockFn := func() (blockNumberAnchor, error) {
+	loadFreshBlockFn := func(prev blockNumberAnchor) (blockNumberAnchor, error) {
 		b.metrics.TotalRPCCalls.Inc()
 		b.metrics.BlockHeaderAsBlockNumberCalls.Inc()
 
@@ -99,9 +102,16 @@ func (b *wrappedBackend) BlockNumber(ctx context.Context) (uint64, error) {
 			b.metrics.TotalRPCErrors.Inc()
 			return blockNumberAnchor{}, errors.New("latest block header unavailable")
 		}
+
+		newNumber := header.Number.Uint64()
+		newTime := time.Unix(int64(header.Time), 0).UTC()
+		averageBlockTime := b.computeAverageBlockTime(prev, newNumber, newTime)
+		b.metrics.AverageBlockTimeSeconds.Set(averageBlockTime.Seconds())
+
 		return blockNumberAnchor{
-			number:    header.Number.Uint64(),
-			timestamp: time.Unix(int64(header.Time), 0).UTC(),
+			number:           newNumber,
+			timestamp:        newTime,
+			averageBlockTime: averageBlockTime,
 		}, nil
 	}
 
@@ -126,10 +136,35 @@ func (b *wrappedBackend) estimatedBlockNumberWithElapsed(anchor blockNumberAncho
 		return anchor.number, 0
 	}
 
-	elapsedBlocks := uint64(now.Sub(anchor.timestamp) / b.blockTime)
+	blockTime := anchor.averageBlockTime
+	if blockTime == 0 {
+		blockTime = b.blockTime
+	}
+
+	elapsedBlocks := uint64(now.Sub(anchor.timestamp) / blockTime)
 	return anchor.number + elapsedBlocks, elapsedBlocks
 }
 
+// computeAverageBlockTime returns the observed block time between prev and the
+// freshly loaded header. Falls back to the configured block time when prev is
+// unset, the block number did not increase, or the header timestamp did not
+// strictly increase.
+func (b *wrappedBackend) computeAverageBlockTime(prev blockNumberAnchor, newNumber uint64, newTime time.Time) time.Duration {
+	if prev.number == 0 || newNumber <= prev.number || !newTime.After(prev.timestamp) {
+		return b.blockTime
+	}
+
+	elapsed := newTime.Sub(prev.timestamp)
+	blocks := newNumber - prev.number
+
+	averageBlockTime := elapsed / time.Duration(blocks)
+	if averageBlockTime > maxAverageBlockTime {
+		return maxAverageBlockTime
+	}
+
+	return averageBlockTime
+}
+
 func (b *wrappedBackend) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
 	b.metrics.TotalRPCCalls.Inc()
 	b.metrics.BlockHeaderCalls.Inc()