Add node health, version, and identity RPC methods

cmd/mithril/node/node.go

@@ -2600,6 +2600,7 @@ postBootstrap:
 				klog.Fatalf("unable to load RPC epoch rewards: %v", err)
 			}
 		}
+		rpcServer.SetIdentity(validatorIdentityPubkey)
 		rpcServer.Start()
 		mlog.Log.Infof("Started RPC server on port %d", rpcPort)
 	}

pkg/blockstream/block_source.go

@@ -318,15 +318,17 @@ type BlockSource struct {
 	maxInflight int
 
 	// Tip tracking
-	confirmedTip      atomic.Uint64
-	processedTip      atomic.Uint64 // Processed commitment tip (super tip)
-	tipAtSlot         atomic.Uint64 // What slot we had executed when tip was measured
-	lastExecutedSlot  atomic.Uint64 // Replay's consumed frontier, including skips (set by SetLastExecutedSlot)
-	tipSafetyMargin   uint64
-	tipPollInterval   time.Duration
-	lastTipUpdate     atomic.Int64  // Unix timestamp of last successful tip poll
-	tipPollFailures   atomic.Uint64 // Consecutive tip poll failures
-	totalTipPollFails atomic.Uint64 // Total tip poll failures (for stats)
+	confirmedTip          atomic.Uint64
+	observedConfirmedTip  atomic.Uint64 // Latest confirmed slot returned by a configured RPC
+	processedTip          atomic.Uint64 // Processed commitment tip (super tip)
+	tipAtSlot             atomic.Uint64 // What slot we had executed when tip was measured
+	lastExecutedSlot      atomic.Uint64 // Replay's consumed frontier, including skips (set by SetLastExecutedSlot)
+	tipSafetyMargin       uint64
+	tipPollInterval       time.Duration
+	lastTipUpdate         atomic.Int64  // Unix timestamp of last tip update (poll or fetched block)
+	lastObservedTipUpdate atomic.Int64  // Unix timestamp of last successful confirmed-tip RPC poll
+	tipPollFailures       atomic.Uint64 // Consecutive tip poll failures
+	totalTipPollFails     atomic.Uint64 // Total tip poll failures (for stats)
 
 	// Reorder buffer
 	reorderMu     sync.Mutex
@@ -2477,10 +2479,19 @@ func (bs *BlockSource) NotifyBlockStart(slot uint64) {
 // This allows accurate distance calculation: tip - tipAtSlot is precise at measurement time.
 func (bs *BlockSource) updateTipSnapshot(confirmedTip uint64) {
 	slotAtTip := bs.lastExecutedSlot.Load()
+	now := time.Now().Unix()
 
 	bs.confirmedTip.Store(confirmedTip)
 	bs.tipAtSlot.Store(slotAtTip)
-	bs.lastTipUpdate.Store(time.Now().Unix())
+	bs.lastTipUpdate.Store(now)
+	bs.observedConfirmedTip.Store(confirmedTip)
+	bs.lastObservedTipUpdate.Store(now)
+}
+
+// HealthSlot returns a recently observed network tip, not a replay-derived estimate.
+func (bs *BlockSource) HealthSlot() (uint64, bool) {
+	updated := bs.lastObservedTipUpdate.Load()
+	return bs.observedConfirmedTip.Load(), updated > 0 && time.Since(time.Unix(updated, 0)) <= max(30*time.Second, 2*bs.tipPollInterval)
 }
 
 // RefreshTipsForSummary triggers an async refresh of both confirmed and processed tips.
@@ -2527,9 +2538,12 @@ func (bs *BlockSource) RefreshTipsForSummary() {
 
 		// Store results
 		if maxConfirmed > 0 {
+			now := time.Now().Unix()
 			bs.confirmedTip.Store(maxConfirmed)
 			bs.tipAtSlot.Store(slotAtTip)
-			bs.lastTipUpdate.Store(time.Now().Unix())
+			bs.lastTipUpdate.Store(now)
+			bs.observedConfirmedTip.Store(maxConfirmed)
+			bs.lastObservedTipUpdate.Store(now)
 			bs.tipPollFailures.Store(0)
 		}
 		if maxProcessed > 0 {

pkg/blockstream/health_slot_test.go

@@ -0,0 +1,31 @@
+package blockstream
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHealthSlotRequiresFreshNetworkObservation(t *testing.T) {
+	bs := &BlockSource{}
+	_, fresh := bs.HealthSlot()
+	require.False(t, fresh)
+	bs.confirmedTip.Store(2000)
+	bs.lastTipUpdate.Store(time.Now().Unix())
+	_, fresh = bs.HealthSlot()
+	require.False(t, fresh, "a locally advanced tip is not a network observation")
+	bs.updateTipSnapshot(1234)
+	slot, fresh := bs.HealthSlot()
+	require.Equal(t, uint64(1234), slot)
+	require.True(t, fresh)
+	bs.lastObservedTipUpdate.Store(time.Now().Add(-31 * time.Second).Unix())
+	_, fresh = bs.HealthSlot()
+	require.False(t, fresh)
+	bs.tipPollInterval = 20 * time.Second
+	_, fresh = bs.HealthSlot()
+	require.True(t, fresh, "a longer configured polling interval needs a longer freshness window")
+	bs.lastObservedTipUpdate.Store(time.Now().Add(-41 * time.Second).Unix())
+	_, fresh = bs.HealthSlot()
+	require.False(t, fresh)
+}

pkg/replay/alpenglow_switch.go

@@ -314,6 +314,7 @@ func tryInLoopUnwind(
 	epochSchedule *sealevel.SysvarEpochSchedule,
 	currentEpoch uint64,
 	partitionedRewardsInfo *rewards.PartitionedRewardDistributionInfo,
+	health rpcHealthInvalidator,
 ) (*ResumeState, *sealevel.BankSysvars, string) {
 	if tail == nil || sw.Slot == 0 {
 		return nil, nil, unwindFallbackNilTail
@@ -342,6 +343,11 @@ func tryInLoopUnwind(
 		return nil, nil, unwindFallbackVoteStakeDirty
 	}
 
+	// Invalidate before eviction, even if rebuilding the retained bank fails
+	// and the caller must restart from its durable checkpoint.
+	if health != nil {
+		health.InvalidateHealth()
+	}
 	ctx, bankSysvars := tail.unwind(sw.Slot)
 	if ctx == nil && sw.Slot-1 == mithrilState.LastRootedSlot && mithrilState.LastRootedContext != nil {
 		// The durable boundary carries a persisted ResumeContext, but deliberately

pkg/replay/alpenglow_unwind_test.go

@@ -246,7 +246,7 @@ func TestTryInLoopUnwindFallsBackWhenVoteStakeDirty(t *testing.T) {
 	// carries the parent's transaction count so the discarded fork's txs can be
 	// dropped from the running count.
 	resetVoteStakeDirty()
-	rs, bankSysvars, _ := tryInLoopUnwind(sw, newTail(), mithrilState, sched, epoch, nil)
+	rs, bankSysvars, _ := tryInLoopUnwind(sw, newTail(), mithrilState, sched, epoch, nil, nil)
 	require.NotNil(t, rs, "clean unwind should succeed in-loop")
 	require.NotNil(t, bankSysvars, "clean unwind returns its exact parent sysvars")
 	assert.Equal(t, uint64(7), bankSysvars.Slot())
@@ -260,7 +260,7 @@ func TestTryInLoopUnwindFallsBackWhenVoteStakeDirty(t *testing.T) {
 	// A global cache was mutated in the UNWOUND suffix (at the switch slot):
 	// the unwind cannot roll the caches back -> must fall back (nil).
 	markVoteStakeDirty(8)
-	rs, _, reason := tryInLoopUnwind(sw, newTail(), mithrilState, sched, epoch, nil)
+	rs, _, reason := tryInLoopUnwind(sw, newTail(), mithrilState, sched, epoch, nil, nil)
 	assert.Nil(t, rs, "dirty cache in the unwound suffix must force the rooted re-replay fallback")
 	assert.Equal(t, unwindFallbackVoteStakeDirty, reason)
 	resetVoteStakeDirty()
@@ -270,7 +270,7 @@ func TestTryInLoopUnwindFallsBackWhenVoteStakeDirty(t *testing.T) {
 	// cache from durable, which cannot see the retained suffix's writes — the
 	// reload would REGRESS the cache below live account state.
 	markVoteStakeDirty(7) // retained slot; rooted watermark is 0
-	rs, _, reason = tryInLoopUnwind(sw, newTail(), mithrilState, sched, epoch, nil)
+	rs, _, reason = tryInLoopUnwind(sw, newTail(), mithrilState, sched, epoch, nil, nil)
 	assert.Nil(t, rs, "dirty cache in the retained suffix must also force the fallback")
 	assert.Equal(t, unwindFallbackVoteStakeDirty, reason)
 	resetVoteStakeDirty()
@@ -279,7 +279,7 @@ func TestTryInLoopUnwindFallsBackWhenVoteStakeDirty(t *testing.T) {
 	// durable and reload-from-durable is exact again -> fast path allowed.
 	markVoteStakeDirty(7)
 	rootedPast := &state.MithrilState{LastRootedSlot: 7}
-	rs, _, _ = tryInLoopUnwind(sw, newTail(), rootedPast, sched, epoch, nil)
+	rs, _, _ = tryInLoopUnwind(sw, newTail(), rootedPast, sched, epoch, nil, nil)
 	require.NotNil(t, rs, "dirtiness at/below the rooted watermark is durably folded — fast path is safe")
 	resetVoteStakeDirty()
 }
@@ -351,7 +351,7 @@ func TestTryInLoopUnwindGuardMatrix(t *testing.T) {
 // assertUnwindOK asserts the in-RAM unwind proceeds (no fallback reason).
 func assertUnwindOK(t *testing.T, sw *CertifiedSwitch, tail *unrootedTail, ms *state.MithrilState, sched *sealevel.SysvarEpochSchedule, epoch uint64, ri *rewards.PartitionedRewardDistributionInfo) {
 	t.Helper()
-	rs, bankSysvars, reason := tryInLoopUnwind(sw, tail, ms, sched, epoch, ri)
+	rs, bankSysvars, reason := tryInLoopUnwind(sw, tail, ms, sched, epoch, ri, nil)
 	require.NotNil(t, rs, "expected in-RAM unwind to proceed, got fallback %q", reason)
 	require.NotNil(t, bankSysvars)
 	assert.Empty(t, reason)
@@ -360,7 +360,7 @@ func assertUnwindOK(t *testing.T, sw *CertifiedSwitch, tail *unrootedTail, ms *s
 // assertUnwindFallback asserts the unwind falls back (any reason).
 func assertUnwindFallback(t *testing.T, sw *CertifiedSwitch, tail *unrootedTail, ms *state.MithrilState, sched *sealevel.SysvarEpochSchedule, epoch uint64, ri *rewards.PartitionedRewardDistributionInfo) {
 	t.Helper()
-	rs, bankSysvars, reason := tryInLoopUnwind(sw, tail, ms, sched, epoch, ri)
+	rs, bankSysvars, reason := tryInLoopUnwind(sw, tail, ms, sched, epoch, ri, nil)
 	assert.Nil(t, rs)
 	assert.Nil(t, bankSysvars)
 	assert.NotEmpty(t, reason, "fallback must carry a reason for the instrumentation")
@@ -370,7 +370,7 @@ func assertUnwindFallback(t *testing.T, sw *CertifiedSwitch, tail *unrootedTail,
 // instrumented reason operators will see.
 func assertUnwindFallbackReason(t *testing.T, want string, sw *CertifiedSwitch, tail *unrootedTail, ms *state.MithrilState, sched *sealevel.SysvarEpochSchedule, epoch uint64, ri *rewards.PartitionedRewardDistributionInfo) {
 	t.Helper()
-	rs, bankSysvars, reason := tryInLoopUnwind(sw, tail, ms, sched, epoch, ri)
+	rs, bankSysvars, reason := tryInLoopUnwind(sw, tail, ms, sched, epoch, ri, nil)
 	assert.Nil(t, rs)
 	assert.Nil(t, bankSysvars)
 	assert.Equal(t, want, reason)

pkg/replay/block.go

@@ -63,6 +63,10 @@ type epochRewardReadPublisher interface {
 	ResetEpochRewardView()
 }
 
+type rpcHealthInvalidator interface {
+	InvalidateHealth()
+}
+
 // BlockFetchOpts contains options for parallel block fetching
 type BlockFetchOpts struct {
 	MaxRPS          int    // Rate limit (requests per second), 0 = use default
@@ -1767,6 +1771,11 @@ func ReplayBlocks(
 	onCancelWriteState OnCancelWriteState, // callback to write state immediately on cancellation (can be nil)
 ) *ReplayResult {
 	result := &ReplayResult{}
+	health, _ := rpcServer.(rpcHealthInvalidator)
+	if health != nil {
+		health.InvalidateHealth()
+		defer health.InvalidateHealth()
+	}
 	alpenglowMode := consensusOpts != nil && consensusOpts.Alpenglow
 	rewardPublisher, _ := rpcServer.(epochRewardPublisher)
 	rewardReadPublisher, _ := rpcServer.(epochRewardReadPublisher)
@@ -2555,6 +2564,10 @@ func ReplayBlocks(
 	}
 
 	blockStream := blockstream.NewBlockSource(opts)
+	if health, ok := rpcServer.(interface{ SetHealthSlotSource(func() (uint64, bool)) }); ok && isLive {
+		health.SetHealthSlotSource(blockStream.HealthSlot)
+		defer health.SetHealthSlotSource(func() (uint64, bool) { return 0, false })
+	}
 
 	if !isLive {
 		blockStream.DownloadInitialBlocks()
@@ -2642,7 +2655,7 @@ func ReplayBlocks(
 			mlog.Log.Warnf("%v — re-serving the skipped suffix from slot %d; executed bank remains at slot %d", sw, sw.Slot, currentExecutedAnchorSlot())
 			return true
 		}
-		rs, parentBankSysvars, fallbackReason := tryInLoopUnwind(sw, unrootedTailState, mithrilState, epochSchedule, currentEpoch, partitionedRewardsInfo)
+		rs, parentBankSysvars, fallbackReason := tryInLoopUnwind(sw, unrootedTailState, mithrilState, epochSchedule, currentEpoch, partitionedRewardsInfo, health)
 		if rs == nil {
 			windowSwitchFallback++
 			switchFallbackReasons[fallbackReason]++
@@ -3824,6 +3837,11 @@ func ReplayBlocks(
 		rewardReadPublisher.ResetEpochRewardView()
 	}
 
+	// Replay has stopped; cleanup may still wait for folds and worker shutdown.
+	if health != nil {
+		health.InvalidateHealth()
+	}
+
 	// Check if block source stalled (this provides explicit error info)
 	if blockStream.Stalled() && result.Error == nil {
 		result.Error = fmt.Errorf("block fetch stalled - no progress for %v", blockStream.StallTimeout())

pkg/replay/rewards_retirement_test.go

@@ -164,13 +164,13 @@ func TestRewardsRetirementAllowsExactParentUnwind(t *testing.T) {
 	sw := &CertifiedSwitch{Slot: 8}
 	ms := &state.MithrilState{LastRootedSlot: 4}
 	sched := &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 432000}
-	rs, _, reason := tryInLoopUnwind(sw, tail, ms, sched, 0, info)
+	rs, _, reason := tryInLoopUnwind(sw, tail, ms, sched, 0, info, nil)
 	require.Nil(t, rs)
 	require.Equal(t, unwindFallbackRewardsWindow, reason)
 	ms.LastRootedSlot = 5
 	markVoteStakeDirty(5) // completed reward writes are also below the durable root
 	require.True(t, completed.retire(&info, ms.LastRootedSlot))
-	rs, restored, reason := tryInLoopUnwind(sw, tail, ms, sched, 0, info)
+	rs, restored, reason := tryInLoopUnwind(sw, tail, ms, sched, 0, info, nil)
 	require.Empty(t, reason)
 	require.Same(t, bank, restored, "use the surviving bank, never abandoned reward sysvars")
 	want, err := ResumeStateFromRootedContext(parent, nil)

pkg/replay/rpc_health_unwind_test.go

@@ -0,0 +1,61 @@
+package replay
+
+import (
+	"encoding/base64"
+	"testing"
+
+	"github.com/Overclock-Validator/mithril/pkg/accounts"
+	"github.com/Overclock-Validator/mithril/pkg/sealevel"
+	"github.com/Overclock-Validator/mithril/pkg/state"
+	"github.com/mr-tron/base58"
+	"github.com/stretchr/testify/require"
+)
+
+type unwindHealthCheck struct {
+	onInvalidate func()
+}
+
+func (h *unwindHealthCheck) InvalidateHealth() { h.onInvalidate() }
+
+func TestTryInLoopUnwindInvalidatesHealthBeforeMutation(t *testing.T) {
+	resetVoteStakeDirty()
+	t.Cleanup(resetVoteStakeDirty)
+	for _, missingSysvars := range []bool{false, true} {
+		t.Run(map[bool]string{false: "retained parent", true: "checkpoint fallback"}[missingSysvars], func(t *testing.T) {
+			tail := newUnrootedTail(&fakeDurable{}, &fakeCommitter{durable: accounts.NewMemAccounts()}, 512, 1, "")
+			parent := &state.ResumeContext{
+				Slot: 7, Bankhash: base58.Encode(make([]byte, 32)),
+				AcctsLtHash: base64.StdEncoding.EncodeToString(make([]byte, 2048)),
+			}
+			account := testAccount(1, 71)
+			tail.Add(7, []*accounts.Account{account}, testHashBytes(7))
+			var sysvars *sealevel.BankSysvars
+			if !missingSysvars {
+				sysvars = testUnwindBankSysvars(t, 7, 700)
+			}
+			tail.SetContext(7, parent, sysvars)
+			tail.Add(8, []*accounts.Account{testAccount(1, 81)}, testHashBytes(8))
+			invalidated := false
+			health := &unwindHealthCheck{onInvalidate: func() {
+				before, err := tail.GetAccount(0, account.Key)
+				require.NoError(t, err)
+				require.Equal(t, uint64(81), before.Lamports, "health must be invalidated before discarded state is evicted")
+				invalidated = true
+			}}
+			schedule := &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 432000}
+			resumed, _, reason := tryInLoopUnwind(&CertifiedSwitch{Slot: 8}, tail, &state.MithrilState{}, schedule, 0, nil, health)
+			after, err := tail.GetAccount(0, account.Key)
+			require.NoError(t, err)
+			require.Equal(t, uint64(71), after.Lamports, "the real unwind restores the retained parent's account")
+			if missingSysvars {
+				require.Nil(t, resumed)
+				require.Equal(t, unwindFallbackMissingSysvars, reason)
+			} else {
+				require.NotNil(t, resumed)
+				require.Equal(t, uint64(7), resumed.ParentSlot)
+				require.Empty(t, reason)
+			}
+			require.True(t, invalidated, "discarding the RPC bank must invalidate health, including failed reconstruction")
+		})
+	}
+}

pkg/rpcserver/errors.go

@@ -13,6 +13,8 @@ const (
 	rpcCodeInvalidParams jsonrpc.ErrorCode = -32602
 	// -32002 matches Agave's SendTransactionPreflightFailure.
 	rpcCodeSendTransactionPreflightFailure jsonrpc.ErrorCode = -32002
+	// -32005 is the Solana RPC NodeUnhealthy error code.
+	rpcCodeNodeUnhealthy jsonrpc.ErrorCode = -32005
 	// -32016 is Agave's reserved code for MinContextSlotNotReached.
 	rpcCodeMinContextSlotNotReached jsonrpc.ErrorCode = -32016
 	rpcCodeBlockNotAvailable        jsonrpc.ErrorCode = -32004
@@ -97,6 +99,49 @@ func (e *EpochRewardHistoryUnavailableError) FromJSONRPCError(rpcErr jsonrpc.JSO
 	return json.Unmarshal(raw, e)
 }
 
+type NodeUnhealthyError struct {
+	NumSlotsBehind *uint64
+}
+
+func (e *NodeUnhealthyError) Error() string {
+	if e.NumSlotsBehind != nil {
+		return fmt.Sprintf("Node is behind by %d slots", *e.NumSlotsBehind)
+	}
+	return "Node is unhealthy"
+}
+
+func (e *NodeUnhealthyError) ToJSONRPCError() (jsonrpc.JSONRPCError, error) {
+	return jsonrpc.JSONRPCError{
+		Code:    rpcCodeNodeUnhealthy,
+		Message: e.Error(),
+		Data: struct {
+			NumSlotsBehind *uint64 `json:"numSlotsBehind"`
+		}{NumSlotsBehind: e.NumSlotsBehind},
+	}, nil
+}
+
+func (e *NodeUnhealthyError) FromJSONRPCError(rpcErr jsonrpc.JSONRPCError) error {
+	if rpcErr.Code != rpcCodeNodeUnhealthy {
+		return fmt.Errorf("unexpected code %d for NodeUnhealthyError", rpcErr.Code)
+	}
+	if rpcErr.Data == nil {
+		e.NumSlotsBehind = nil
+		return nil
+	}
+	raw, err := json.Marshal(rpcErr.Data)
+	if err != nil {
+		return fmt.Errorf("re-encoding NodeUnhealthyError data: %w", err)
+	}
+	var payload struct {
+		NumSlotsBehind *uint64 `json:"numSlotsBehind"`
+	}
+	if err := json.Unmarshal(raw, &payload); err != nil {
+		return fmt.Errorf("decoding NodeUnhealthyError data: %w", err)
+	}
+	e.NumSlotsBehind = payload.NumSlotsBehind
+	return nil
+}
+
 type MinContextSlotNotReachedError struct {
 	ContextSlot uint64
 }
@@ -194,6 +239,7 @@ func rpcErrorRegistry() jsonrpc.Errors {
 	errs := jsonrpc.NewErrors()
 	errs.Register(rpcCodeInvalidParams, new(*InvalidParamsError))
 	errs.Register(rpcCodeSendTransactionPreflightFailure, new(*SendTransactionPreflightFailureError))
+	errs.Register(rpcCodeNodeUnhealthy, new(*NodeUnhealthyError))
 	errs.Register(rpcCodeMinContextSlotNotReached, new(*MinContextSlotNotReachedError))
 	return errs
 }