Add node health, version, and identity RPC methods

cmd/mithril/node/node.go

@@ -2566,6 +2566,7 @@ postBootstrap:
 		klog.Fatalf("invalid port: %d", rpcPort)
 	} else if rpcPort != 0 {
 		rpcServer = rpcserver.NewRpcServer(accountsDb, uint16(rpcPort), epochScheduleFromState(mithrilState), solana.MustHashFromBase58(networkGenesisHash))
+		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
@@ -2474,10 +2476,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.
@@ -2524,9 +2535,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/block.go

@@ -2407,6 +2407,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()

pkg/rpcserver/errors.go

@@ -13,10 +13,55 @@ 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
 )
 
+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
 }
@@ -114,6 +159,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
 }

pkg/rpcserver/errors_test.go

@@ -81,3 +81,40 @@ func TestInvalidParamsError_FromJSONRPCError(t *testing.T) {
 	require.NoError(t, e.FromJSONRPCError(jsonrpc.JSONRPCError{Code: -32602, Message: "x"}))
 	assert.Equal(t, "x", e.Message)
 }
+
+func TestNodeUnhealthyErrorWireShape(t *testing.T) {
+	behind := uint64(129)
+	rpcErr, err := (&NodeUnhealthyError{NumSlotsBehind: &behind}).ToJSONRPCError()
+	require.NoError(t, err)
+	require.Equal(t, jsonrpc.ErrorCode(-32005), rpcErr.Code)
+	require.Equal(t, "Node is behind by 129 slots", rpcErr.Message)
+
+	raw, err := json.Marshal(rpcErr)
+	require.NoError(t, err)
+	assert.Contains(t, string(raw), `"data":{"numSlotsBehind":129}`)
+
+	withoutDistance, err := (&NodeUnhealthyError{}).ToJSONRPCError()
+	require.NoError(t, err)
+	raw, err = json.Marshal(withoutDistance)
+	require.NoError(t, err)
+	assert.Contains(t, string(raw), `"data":{"numSlotsBehind":null}`)
+}
+
+func TestNodeUnhealthyErrorFromJSONRPCError(t *testing.T) {
+	var decoded NodeUnhealthyError
+	require.NoError(t, decoded.FromJSONRPCError(jsonrpc.JSONRPCError{
+		Code: -32005,
+		Data: map[string]interface{}{"numSlotsBehind": float64(42)},
+	}))
+	require.NotNil(t, decoded.NumSlotsBehind)
+	assert.Equal(t, uint64(42), *decoded.NumSlotsBehind)
+
+	require.NoError(t, decoded.FromJSONRPCError(jsonrpc.JSONRPCError{Code: -32005}))
+	assert.Nil(t, decoded.NumSlotsBehind)
+	require.NoError(t, decoded.FromJSONRPCError(jsonrpc.JSONRPCError{
+		Code: -32005,
+		Data: map[string]interface{}{"numSlotsBehind": nil},
+	}))
+	assert.Nil(t, decoded.NumSlotsBehind)
+	assert.Error(t, decoded.FromJSONRPCError(jsonrpc.JSONRPCError{Code: 1}))
+}

pkg/rpcserver/node_health_source_test.go

@@ -0,0 +1,45 @@
+package rpcserver
+
+import (
+	"testing"
+
+	"github.com/Overclock-Validator/mithril/pkg/global"
+	"github.com/Overclock-Validator/mithril/pkg/sealevel"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNodeHealthUsesObservedNetworkTip(t *testing.T) {
+	local := global.WallClockSlot()
+	server := &RpcServer{}
+	server.SetSlotCtx(&sealevel.SlotCtx{Slot: local})
+	for _, test := range []struct {
+		name        string
+		gap         uint64
+		fresh       bool
+		wantHealthy bool
+	}{
+		{"caught up", 0, true, true},
+		{"catching up", 1000, true, false},
+		{"stale network tip", 0, false, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			server.SetHealthSlotSource(func() (uint64, bool) { return local + test.gap, test.fresh })
+			value, err := server.GetHealth(t.Context(), nil)
+			if test.wantHealthy {
+				require.NoError(t, err)
+				require.Equal(t, "ok", value)
+			} else {
+				var unhealthy *NodeUnhealthyError
+				require.ErrorAs(t, err, &unhealthy)
+			}
+		})
+	}
+}
+
+func TestNodeHealthRequiresNetworkTipSource(t *testing.T) {
+	server := &RpcServer{}
+	server.SetSlotCtx(&sealevel.SlotCtx{Slot: global.WallClockSlot()})
+	_, err := server.GetHealth(t.Context(), nil)
+	var unhealthy *NodeUnhealthyError
+	require.ErrorAs(t, err, &unhealthy)
+}

pkg/rpcserver/node_info.go

@@ -0,0 +1,86 @@
+package rpcserver
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/Overclock-Validator/mithril/pkg/version"
+	"github.com/filecoin-project/go-jsonrpc"
+)
+
+// healthCheckSlotDistance is the default Solana RPC health-check distance.
+const healthCheckSlotDistance = 128
+
+type GetVersionResp struct {
+	SolanaCore string  `json:"solana-core"`
+	FeatureSet *uint32 `json:"feature-set"`
+}
+
+type GetIdentityResp struct {
+	Identity string `json:"identity"`
+}
+
+func (rpcServer *RpcServer) GetHealth(ctx context.Context, p jsonrpc.RawParams) (string, error) {
+	if err := validateNoParams(ctx, "getHealth", p); err != nil {
+		return "", err
+	}
+
+	rpcServer.slotCtxMu.RLock()
+	slotCtx, healthSlot := rpcServer.slotCtx, rpcServer.healthSlot
+	rpcServer.slotCtxMu.RUnlock()
+	if slotCtx == nil || healthSlot == nil {
+		return nodeHealth(false, 0, 0)
+	}
+	clusterSlot, fresh := healthSlot()
+	return nodeHealth(fresh, slotCtx.Slot, clusterSlot)
+}
+
+// SetHealthSlotSource uses the block source's cached network tip without polling per request.
+func (rpcServer *RpcServer) SetHealthSlotSource(source func() (uint64, bool)) {
+	rpcServer.slotCtxMu.Lock()
+	rpcServer.healthSlot = source
+	rpcServer.slotCtxMu.Unlock()
+}
+
+func nodeHealth(ready bool, localSlot, clusterSlot uint64) (string, error) {
+	if !ready {
+		return "", &NodeUnhealthyError{}
+	}
+	if clusterSlot > localSlot && clusterSlot-localSlot > healthCheckSlotDistance {
+		behind := clusterSlot - localSlot
+		return "", &NodeUnhealthyError{NumSlotsBehind: &behind}
+	}
+	return "ok", nil
+}
+
+func (rpcServer *RpcServer) GetVersion(ctx context.Context, p jsonrpc.RawParams) (GetVersionResp, error) {
+	if err := validateNoParams(ctx, "getVersion", p); err != nil {
+		return GetVersionResp{}, err
+	}
+	return GetVersionResp{SolanaCore: version.Version}, nil
+}
+
+func (rpcServer *RpcServer) GetIdentity(ctx context.Context, p jsonrpc.RawParams) (GetIdentityResp, error) {
+	if err := validateNoParams(ctx, "getIdentity", p); err != nil {
+		return GetIdentityResp{}, err
+	}
+	if rpcServer.identity == "" {
+		return GetIdentityResp{}, errors.New("validator identity is not configured")
+	}
+	return GetIdentityResp{Identity: rpcServer.identity}, nil
+}
+
+func validateNoParams(ctx context.Context, method string, p jsonrpc.RawParams) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if len(p) == 0 {
+		return nil
+	}
+	params, err := jsonrpc.DecodeParams[[]json.RawMessage](p)
+	if err != nil || len(params) != 0 {
+		return &InvalidParamsError{Message: method + " does not accept parameters"}
+	}
+	return nil
+}