forkable: add WithFinalizedBlockNumMetric option

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `forkable.WithFinalizedBlockNumMetric`: new option reporting, on the live path, the LIB number of the block that just became head. Takes a `forkable.Uint64Metric` interface so consumers can provide their own metric implementation.
 - `DefaultMergedBlocksBundleSize`: new package variable (default `100`) controlling the number of blocks per merged-blocks file assumed by readers when no explicit size is given. Like `GetProtocolFirstStreamableBlock`, it is meant to be set once at process startup.
 - `stream.WithMergedBlocksBundleSize`: new `stream` option to set the merged-blocks bundle size for a single stream (overrides the process-wide default; used by substreams tier2 which serves multiple chains at once).
 - `FileSource` now fails fast with a clear error when a merged-blocks file contains a block beyond the configured bundle size (store files bigger than the configured size).

forkable/forkable.go

@@ -56,6 +56,7 @@ type Forkable struct {
 	metricsHeadBlockNum       *dmetrics.HeadBlockNum
 	metricsHeadTimeDrift      *dmetrics.HeadTimeDrift
 	metricsRelativeBlockDrift *dmetrics.HeadBlockRelativeTime
+	metricsFinalizedBlockNum  Uint64Metric
 
 	lastLongestChain []*Block
 }
@@ -781,6 +782,9 @@ func (p *Forkable) ProcessBlock(blk *pbbstream.Block, obj any) error {
 		if p.metricsRelativeBlockDrift != nil {
 			p.metricsRelativeBlockDrift.SetLastBlock(blk.Time())
 		}
+		if p.metricsFinalizedBlockNum != nil {
+			p.metricsFinalizedBlockNum.SetUint64(blk.LibNum)
+		}
 	}
 
 	if p.lastBlockSent == nil {

forkable/metrics_test.go

@@ -0,0 +1,40 @@
+package forkable
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type testUint64Metric struct {
+	value uint64
+	set   bool
+}
+
+func (t *testUint64Metric) SetUint64(value uint64) {
+	t.value = value
+	t.set = true
+}
+
+func TestForkable_FinalizedBlockNumMetric(t *testing.T) {
+	finalized := &testUint64Metric{}
+
+	p := New(&testForkableSink{}, WithFinalizedBlockNumMetric(finalized), WithExclusiveLIB(bRef("00000003a")))
+	p.SetLiveMetrics()
+
+	require.NoError(t, p.ProcessBlock(tb("00000004a", "00000003a", 3), nil))
+	assert.Equal(t, uint64(3), finalized.value)
+
+	require.NoError(t, p.ProcessBlock(tb("00000005a", "00000004a", 4), nil))
+	assert.Equal(t, uint64(4), finalized.value)
+}
+
+func TestForkable_FinalizedBlockNumMetricNotSetWithoutLiveMetrics(t *testing.T) {
+	finalized := &testUint64Metric{}
+
+	p := New(&testForkableSink{}, WithFinalizedBlockNumMetric(finalized), WithExclusiveLIB(bRef("00000003a")))
+
+	require.NoError(t, p.ProcessBlock(tb("00000004a", "00000003a", 3), nil))
+	assert.False(t, finalized.set, "finalized block num metric must only be updated on the live path")
+}

forkable/options.go

@@ -42,6 +42,22 @@ func WithMetrics(
 	}
 }
 
+// Uint64Metric is the minimal interface a metric must satisfy to be updated by the
+// forkable. It exists so that consumers can provide their own metric implementation
+// without bstream having to depend on it.
+type Uint64Metric interface {
+	SetUint64(value uint64)
+}
+
+// WithFinalizedBlockNumMetric reports, on the live path, the LIB number of the block
+// that just became head. Combined with the head block number metric, it tells how far
+// behind finality the head is.
+func WithFinalizedBlockNumMetric(finalizedBlockNum Uint64Metric) Option {
+	return func(f *Forkable) {
+		f.metricsFinalizedBlockNum = finalizedBlockNum
+	}
+}
+
 func WithWarnOnUnlinkableBlocks(count int) Option {
 	return func(f *Forkable) {
 		f.warnOnUnlinkableBlocksCount = count