fix(node,store): take the full-table scans off the dispatch loop and repair finalization pruning

cmd/gean/main.go

@@ -63,6 +63,15 @@ func run(cfg config) error {
 		return err
 	}
 
+	// Recover the stored-block high-water mark before any duty runs. Doing it
+	// here rather than lazily keeps the full-table scan off the dispatch loop
+	// entirely, and surfaces a read failure at startup instead of leaving the
+	// duty gate to act on an understated mark.
+	if err := s.SeedMaxStoredBlockSlot(); err != nil {
+		logger.Error(logger.Node, "seed max stored block slot: %v", err)
+		return err
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -99,5 +108,11 @@ func run(cfg config) error {
 	logger.Info(logger.Node, "gean started: api=%s metrics=%s aggregator=%v", apiAddr, metricsAddr, cfg.IsAggregator)
 
 	waitForShutdown(cancel)
+
+	// Join the storage-size sampler before the deferred backend.Close runs.
+	// Cancellation alone is not enough: a sampler mid-round when the database
+	// closes calls into a closed Pebble instance, which panics. Other workers
+	// that read storage are still not joined — a pre-existing gap.
+	n.WaitForStorageWorkers()
 	return nil
 }

internal/blockbuilder/build_test.go

@@ -178,7 +178,7 @@ func TestBuildBlockReportsMismatchedPayloadRoot(t *testing.T) {
 		Slot:            1,
 		ProposerIndex:   0,
 		ParentRoot:      parentRoot,
-		KnownBlockRoots: map[[32]byte]bool{parentRoot: true},
+		KnownBlockRoots: RootSet{parentRoot: true},
 		Payloads: []AttestationPayload{{
 			DataRoot: [32]byte{0xee},
 			Data:     data,
@@ -218,7 +218,7 @@ func TestBuildBlockReportsSkippedPayloadIssues(t *testing.T) {
 		Slot:            3,
 		ProposerIndex:   0,
 		ParentRoot:      parentRoot,
-		KnownBlockRoots: map[[32]byte]bool{parentRoot: true},
+		KnownBlockRoots: RootSet{parentRoot: true},
 		Payloads: []AttestationPayload{
 			{DataRoot: dataRoot, Data: data, Proofs: []*types.SingleMessageAggregate{mockProof([]uint64{0})}},
 			{DataRoot: staleRoot, Data: &staleSourceVote, Proofs: []*types.SingleMessageAggregate{mockProof([]uint64{0})}},
@@ -255,7 +255,7 @@ func TestBuildBlockRecordsProofMergeFallback(t *testing.T) {
 		Slot:            3,
 		ProposerIndex:   0,
 		ParentRoot:      parentRoot,
-		KnownBlockRoots: map[[32]byte]bool{parentRoot: true},
+		KnownBlockRoots: RootSet{parentRoot: true},
 		Payloads: []AttestationPayload{{
 			DataRoot: dataRoot,
 			Data:     data,
@@ -290,7 +290,7 @@ func TestPlanAttestationsUsesPostHeaderState(t *testing.T) {
 		Slot:            3,
 		ProposerIndex:   0,
 		ParentRoot:      parentRoot,
-		KnownBlockRoots: map[[32]byte]bool{parentRoot: true},
+		KnownBlockRoots: RootSet{parentRoot: true},
 		Payloads: []AttestationPayload{{
 			DataRoot: dataRoot,
 			Data:     data,
@@ -358,7 +358,7 @@ func TestPlanAttestationsCarriesTrialState(t *testing.T) {
 		Slot:            3,
 		ProposerIndex:   0,
 		ParentRoot:      parentRoot,
-		KnownBlockRoots: map[[32]byte]bool{root1: true, parentRoot: true},
+		KnownBlockRoots: RootSet{root1: true, parentRoot: true},
 		Payloads: []AttestationPayload{
 			{DataRoot: firstRoot, Data: first, Proofs: []*types.SingleMessageAggregate{mockProof([]uint64{0})}},
 			{DataRoot: secondRoot, Data: second, Proofs: []*types.SingleMessageAggregate{mockProof([]uint64{0})}},
@@ -426,7 +426,7 @@ func TestPlanAttestationsContinuesWhenJustifiedSlotsChange(t *testing.T) {
 		Slot:            7,
 		ProposerIndex:   0,
 		ParentRoot:      parentRoot,
-		KnownBlockRoots: map[[32]byte]bool{roots[2]: true, parentRoot: true},
+		KnownBlockRoots: RootSet{roots[2]: true, parentRoot: true},
 		Payloads: []AttestationPayload{
 			{DataRoot: firstRoot, Data: first, Proofs: []*types.SingleMessageAggregate{mockProof([]uint64{0})}},
 			{DataRoot: secondRoot, Data: second, Proofs: []*types.SingleMessageAggregate{mockProof([]uint64{0})}},
@@ -502,7 +502,7 @@ func TestPlanAttestationsDoesNotReportSkippedPayloadsWhenFull(t *testing.T) {
 		Slot:            2,
 		ProposerIndex:   0,
 		ParentRoot:      parentRoot,
-		KnownBlockRoots: map[[32]byte]bool{parentRoot: true},
+		KnownBlockRoots: RootSet{parentRoot: true},
 		Payloads:        payloads,
 	})
 	if err != nil {

internal/blockbuilder/model.go

@@ -11,12 +11,33 @@ type AttestationPayload struct {
 	Proofs   []*types.SingleMessageAggregate
 }
 
-type KnownRoots map[[32]byte]bool
+// KnownRoots answers whether a block root is one this node has stored.
+//
+// The builder only ever asks membership, a handful of times per proposal, so
+// this is a predicate rather than a set. It used to be a map that the proposal
+// path filled by scanning every key in TableBlockHeaders and allocating an
+// entry per root — tens of thousands of them on a mature chain, rebuilt for
+// every block produced, on the tick loop.
+type KnownRoots interface {
+	Contains(root [32]byte) bool
+}
+
+// RootSet is the in-memory implementation, used by tests and by any caller that
+// genuinely holds the whole set already.
+type RootSet map[[32]byte]bool
 
-func (roots KnownRoots) Contains(root [32]byte) bool {
+func (roots RootSet) Contains(root [32]byte) bool {
 	return roots[root]
 }
 
+// KnownRootsFunc adapts a plain lookup — a point read against storage, say —
+// into a KnownRoots.
+type KnownRootsFunc func(root [32]byte) bool
+
+func (f KnownRootsFunc) Contains(root [32]byte) bool {
+	return f(root)
+}
+
 type Input struct {
 	HeadState       *types.State
 	Slot            uint64

internal/blockbuilder/payloads.go

@@ -78,7 +78,11 @@ func validatePayload(payload AttestationPayload) error {
 
 func payloadBuildIssue(state *types.State, knownRoots KnownRoots, payload AttestationPayload) error {
 	data := payload.Data
-	if !knownRoots.Contains(data.Head.Root) {
+	// A nil KnownRoots knows nothing, matching the nil-map behaviour this
+	// replaced. validateInput already rejects nil whenever there are payloads to
+	// build, so this is the belt to that braces — but an interface, unlike a map,
+	// panics rather than returning false when nil, so it has to be explicit.
+	if knownRoots == nil || !knownRoots.Contains(data.Head.Root) {
 		return errPayloadHeadUnknown(data.Head.Root)
 	}
 	// Only source votes from the chain's current justified checkpoint: older sources

internal/blockbuilder/payloads_test.go

@@ -188,7 +188,7 @@ func TestPayloadBuildIssueUsesTransitionVoteRules(t *testing.T) {
 		Data:     data,
 		Proofs:   []*types.SingleMessageAggregate{mockProof([]uint64{0})},
 	}
-	knownRoots := map[[32]byte]bool{parentRoot: true}
+	knownRoots := RootSet{parentRoot: true}
 	if err := payloadBuildIssue(workingState, knownRoots, payload); err != nil {
 		t.Fatalf("expected payload to be buildable, got %v", err)
 	}
@@ -204,7 +204,7 @@ func TestPayloadBuildIssueUsesTransitionVoteRules(t *testing.T) {
 	offHead := *data
 	forkHead := [32]byte{0xfe}
 	offHead.Head = &types.Checkpoint{Slot: data.Head.Slot, Root: forkHead}
-	if err := payloadBuildIssue(workingState, map[[32]byte]bool{parentRoot: true, forkHead: true},
+	if err := payloadBuildIssue(workingState, RootSet{parentRoot: true, forkHead: true},
 		AttestationPayload{DataRoot: dataRoot, Data: &offHead, Proofs: payload.Proofs}); !errors.Is(err, ErrPayloadHeadOffChain) {
 		t.Fatalf("payload build issue=%v, want ErrPayloadHeadOffChain", err)
 	} else if !IsExpectedSkip(err) {
@@ -279,7 +279,7 @@ func TestPayloadBuildIssueSkipsStaleSource(t *testing.T) {
 		Data:     &staleSource,
 		Proofs:   []*types.SingleMessageAggregate{mockProof([]uint64{0})},
 	}
-	err = payloadBuildIssue(workingState, map[[32]byte]bool{parentRoot: true}, payload)
+	err = payloadBuildIssue(workingState, RootSet{parentRoot: true}, payload)
 	if !errors.Is(err, ErrPayloadSourceNotCurrentJustified) {
 		t.Fatalf("payload build issue=%v, want ErrPayloadSourceNotCurrentJustified", err)
 	}
@@ -301,7 +301,7 @@ func TestPayloadBuildIssueAllowsGenesisSelfVote(t *testing.T) {
 		Target: &types.Checkpoint{Root: root},
 	}
 	dataRoot := hashAttestationData(t, data)
-	err := payloadBuildIssue(state, map[[32]byte]bool{root: true}, AttestationPayload{
+	err := payloadBuildIssue(state, RootSet{root: true}, AttestationPayload{
 		DataRoot: dataRoot,
 		Data:     data,
 		Proofs:   []*types.SingleMessageAggregate{mockProof([]uint64{0})},

internal/blockprocessor/persist.go

@@ -90,6 +90,10 @@ func persistBlock(
 	if err := wb.Commit(); err != nil {
 		return fmt.Errorf("persist block: commit: %w", err)
 	}
+	// The header is now stored, so the duty gate's stored-block high-water mark
+	// has to see it. This is the import path's equivalent of what
+	// ConsensusStore.PutBlockHeader does for pending blocks.
+	s.ObserveStoredBlockSlot(block.Slot)
 	return nil
 }
 

internal/metrics/gauges.go

@@ -12,6 +12,14 @@ var (
 	metricCurrentSlot = promauto.NewGauge(prometheus.GaugeOpts{
 		Name: "lean_current_slot", Help: "Current slot from wall clock",
 	})
+	// metricTickAge is the stall detector. lean_tick_interval_duration_seconds is
+	// a histogram observed *inside* onTick, so it records nothing at all while the
+	// dispatch loop is blocked — the failure it should report makes it go quiet
+	// rather than spike, and its top bucket is 1.6s besides. This gauge is written
+	// from a separate goroutine and keeps climbing for as long as the loop is stuck.
+	metricTickAge = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "lean_tick_last_age_seconds", Help: "Seconds since the dispatch loop last began a tick",
+	})
 	metricSafeTargetSlot = promauto.NewGauge(prometheus.GaugeOpts{
 		Name: "lean_safe_target_slot", Help: "Safe target slot for attestation",
 	})

internal/metrics/histograms.go

@@ -121,6 +121,15 @@ var (
 		Help:    "Elapsed time between clock ticks in seconds",
 		Buckets: []float64{0.4, 0.6, 0.75, 0.8, 0.805, 0.81, 0.815, 0.82, 0.825, 0.85, 0.9, 1.0, 1.2, 1.6},
 	})
+	// metricDispatchEventDuration times each case of the dispatch select, so a
+	// slow handler can be attributed rather than only observed as a late tick.
+	// Buckets run well past a slot: the point is to size a stall, and the
+	// tick-interval histogram's 1.6s ceiling could not.
+	metricDispatchEventDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "lean_dispatch_event_duration_seconds",
+		Help:    "Time the dispatch loop spent handling one event, by event kind",
+		Buckets: []float64{0.001, 0.01, 0.05, 0.1, 0.25, 0.5, 0.8, 1.6, 4, 10, 30, 120, 600},
+	}, []string{"event"})
 	metricProvingDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
 		Name:    "lean_proving_duration_seconds",
 		Help:    "Recursive proof operation duration",

internal/metrics/labels.go

@@ -9,15 +9,13 @@ var syncStatusLabels = []string{"idle", "syncing", "synced", unknownLabel}
 
 const (
 	AggregatorSkipNotAggregator = "not_aggregator"
-	AggregatorSkipNotSynced     = "not_synced"
 	AggregatorSkipMissingState  = "missing_state"
 	AggregatorSkipSpawnFailed   = "spawn_failed"
 	AggregatorSkipOther         = "other"
 )
 
 var aggregatorSkipReasons = []string{
 	AggregatorSkipNotAggregator,
-	AggregatorSkipNotSynced,
 	AggregatorSkipMissingState,
 	AggregatorSkipSpawnFailed,
 	AggregatorSkipOther,

internal/metrics/metrics_test.go

@@ -27,7 +27,6 @@ func TestCountCounterWrappersIgnoreNonPositiveValues(t *testing.T) {
 func TestIncAggregatorSkippedUsesBoundedReasons(t *testing.T) {
 	reasons := []string{
 		AggregatorSkipNotAggregator,
-		AggregatorSkipNotSynced,
 		AggregatorSkipMissingState,
 		AggregatorSkipSpawnFailed,
 		AggregatorSkipOther,

internal/metrics/observe.go

@@ -25,6 +25,15 @@ func ObserveForkChoiceReorgDepth(depth float64) {
 func ObserveTickIntervalDuration(seconds float64) {
 	observeNonNegative(metricTickIntervalDuration, seconds)
 }
+
+// ObserveDispatchEvent records how long one dispatch-loop event took.
+func ObserveDispatchEvent(event string, seconds float64) {
+	if seconds < 0 {
+		return
+	}
+	metricDispatchEventDuration.WithLabelValues(event).Observe(seconds)
+}
+
 func ObserveSTFTime(seconds float64)      { observeNonNegative(metricSTFTime, seconds) }
 func ObserveSTFSlotsTime(seconds float64) { observeNonNegative(metricSTFSlotsTime, seconds) }
 func ObserveSTFBlockTime(seconds float64) { observeNonNegative(metricSTFBlockTime, seconds) }

internal/metrics/set.go

@@ -6,6 +6,7 @@ func SetNodeInfo(name, version string) {
 func SetNodeStartTime(t float64)      { setNonNegative(metricNodeStartTime, t) }
 func SetHeadSlot(s uint64)            { metricHeadSlot.Set(float64(s)) }
 func SetCurrentSlot(s uint64)         { metricCurrentSlot.Set(float64(s)) }
+func SetTickAge(seconds float64)      { metricTickAge.Set(seconds) }
 func SetSafeTargetSlot(s uint64)      { metricSafeTargetSlot.Set(float64(s)) }
 func SetLatestJustifiedSlot(s uint64) { metricLatestJustifiedSlot.Set(float64(s)) }
 func SetLatestFinalizedSlot(s uint64) { metricLatestFinalizedSlot.Set(float64(s)) }

internal/node/dispatch.go

@@ -5,8 +5,19 @@ import (
 	"time"
 
 	"github.com/geanlabs/gean/internal/logger"
+	"github.com/geanlabs/gean/internal/metrics"
 )
 
+// timeEvent records how long one dispatch-loop event took. Every case runs on
+// the single goroutine that also keeps the slot clock, so an unattributed slow
+// handler here shows up only as a late tick — which is exactly how a set of
+// full-table storage scans went unnoticed until they were costing minutes.
+func timeEvent(event string, fn func()) {
+	start := time.Now()
+	fn()
+	metrics.ObserveDispatchEvent(event, time.Since(start).Seconds())
+}
+
 func (e *Engine) dispatch(ctx context.Context, ticks <-chan time.Time) {
 	for {
 		select {
@@ -15,19 +26,19 @@ func (e *Engine) dispatch(ctx context.Context, ticks <-chan time.Time) {
 			return
 
 		case <-ticks:
-			e.onTick()
+			timeEvent("tick", e.onTick)
 
 		case <-e.EarlyAggregateCh:
-			e.maybeEarlyAggregate(uint64(time.Now().UnixMilli()))
+			timeEvent("early_aggregate", func() { e.maybeEarlyAggregate(uint64(time.Now().UnixMilli())) })
 
 		case block := <-e.BlockCh:
-			e.onBlock(block)
+			timeEvent("block", func() { e.onBlock(block) })
 
 		case result := <-e.ProposalResultCh:
-			e.acceptProposal(ctx, result)
+			timeEvent("proposal_result", func() { e.acceptProposal(ctx, result) })
 
 		case root := <-e.FailedRootCh:
-			e.onFailedRoot(root)
+			timeEvent("failed_root", func() { e.onFailedRoot(root) })
 		}
 	}
 }

internal/node/engine.go

@@ -2,6 +2,7 @@ package node
 
 import (
 	"context"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -68,8 +69,24 @@ type Engine struct {
 	RecoveryCh            chan *types.SignedBlock
 	ProvingGate           *proving.Gate
 
+	// storageWorkers tracks the storage-size sampler so shutdown can join it
+	// before the database is closed: a sampler still running after Close calls
+	// into a closed Pebble instance, which panics rather than erroring.
+	//
+	// Scope is deliberately narrow. Other workers read storage too — the
+	// aggregation, proposal, recovery and attestation workers, and the fetch
+	// batcher — and none of them is joined either. That is a pre-existing
+	// shutdown weakness, not one this sampler introduced, and closing it means
+	// deciding how long shutdown may block on in-flight proving work. Tracked
+	// separately; do not read this WaitGroup as covering them.
+	storageWorkers sync.WaitGroup
+
 	lastTick time.Time
 
+	// lastTickMs mirrors lastTick for the stall sampler, which runs on its own
+	// goroutine precisely so it still reports while the dispatch loop is blocked.
+	lastTickMs atomic.Int64
+
 	warnedMissingJustified [32]byte
 
 	// maxSeenGossipSlot is the highest plausible slot heard on gossip, whether
@@ -142,6 +159,16 @@ func New(
 	return e
 }
 
+// WaitForStorageWorkers blocks until the storage-size sampler has returned.
+// Callers must invoke it after cancelling the context and before closing the
+// backend.
+//
+// It does not cover every storage-reading goroutine — see the storageWorkers
+// field for what is and is not tracked.
+func (e *Engine) WaitForStorageWorkers() {
+	e.storageWorkers.Wait()
+}
+
 func (e *Engine) Run(ctx context.Context) {
 	e.initMetrics()