From 5cf941d209229b686a57f51950c44fb01995d7c9 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Tue, 22 Sep 2026 13:57:57 +0530 Subject: [PATCH 1/6] feat(rpc): add exporter node information methods --- cmd/mithril/node/node.go | 1 + pkg/rpcserver/errors.go | 46 +++++++++++++++++++++++ pkg/rpcserver/node_info.go | 77 ++++++++++++++++++++++++++++++++++++++ pkg/rpcserver/rpcserver.go | 8 ++++ 4 files changed, 132 insertions(+) create mode 100644 pkg/rpcserver/node_info.go diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index c452d494e..c44a70d05 100644 --- a/cmd/mithril/node/node.go +++ b/cmd/mithril/node/node.go @@ -2517,6 +2517,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) } diff --git a/pkg/rpcserver/errors.go b/pkg/rpcserver/errors.go index ee2a1654f..8ddf704ea 100644 --- a/pkg/rpcserver/errors.go +++ b/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 } diff --git a/pkg/rpcserver/node_info.go b/pkg/rpcserver/node_info.go new file mode 100644 index 000000000..23702b3cc --- /dev/null +++ b/pkg/rpcserver/node_info.go @@ -0,0 +1,77 @@ +package rpcserver + +import ( + "context" + "encoding/json" + "errors" + + "github.com/Overclock-Validator/mithril/pkg/global" + "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 + } + + slotCtx := rpcServer.getSlotCtx() + if slotCtx == nil { + return nodeHealth(false, 0, 0) + } + return nodeHealth(true, slotCtx.Slot, global.WallClockSlot()) +} + +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 +} diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 8a3dd7afe..8f6f8547b 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -31,6 +31,7 @@ type RpcServer struct { slotCtx *sealevel.SlotCtx slotCtxMu sync.RWMutex genesisHash string + identity string leaderTPUCacheMu sync.RWMutex leaderTPUByIdentity map[solana.PublicKey]tpuEndpoint @@ -53,7 +54,10 @@ var supportedRPCMethods = map[string]struct{}{ "getBlockHeight": {}, "getEpochInfo": {}, "getGenesisHash": {}, + "getHealth": {}, + "getIdentity": {}, "getLatestBlockhash": {}, + "getVersion": {}, "sendTransaction": {}, "simulateTransaction": {}, } @@ -112,6 +116,10 @@ func (rpcServer *RpcServer) SetSlotCtx(slotCtx *sealevel.SlotCtx) { rpcServer.slotCtxMu.Unlock() } +func (rpcServer *RpcServer) SetIdentity(identity string) { + rpcServer.identity = identity +} + func (rpcServer *RpcServer) getSlotCtx() *sealevel.SlotCtx { rpcServer.slotCtxMu.RLock() defer rpcServer.slotCtxMu.RUnlock() From 72cb816853cc7125c82b1664d5645c1b721aebb2 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Tue, 22 Sep 2026 13:57:57 +0530 Subject: [PATCH 2/6] test(rpc): cover exporter node information methods --- pkg/rpcserver/errors_test.go | 37 ++++++++ pkg/rpcserver/node_info_test.go | 153 ++++++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+) create mode 100644 pkg/rpcserver/node_info_test.go diff --git a/pkg/rpcserver/errors_test.go b/pkg/rpcserver/errors_test.go index 6967eb7d8..d6a203659 100644 --- a/pkg/rpcserver/errors_test.go +++ b/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})) +} diff --git a/pkg/rpcserver/node_info_test.go b/pkg/rpcserver/node_info_test.go new file mode 100644 index 000000000..e79539cee --- /dev/null +++ b/pkg/rpcserver/node_info_test.go @@ -0,0 +1,153 @@ +package rpcserver + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/Overclock-Validator/mithril/pkg/version" + "github.com/filecoin-project/go-jsonrpc" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestNodeInfoHTTP(t *testing.T) { + oldVersion := version.Version + version.Version = "1.2.3-test" + t.Cleanup(func() { version.Version = oldVersion }) + + server := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{}, solana.Hash(sha256.Sum256([]byte("cluster")))) + require.NoError(t, server.listener.Close()) + server.SetIdentity("2r1F4iWqVcb8M1DbAjQuFpebkQHY9hcVU4WuW2DJBppN") + server.SetSlotCtx(&sealevel.SlotCtx{Slot: global.WallClockSlot()}) + endpoint := httptest.NewServer(server) + t.Cleanup(endpoint.Close) + + for _, test := range []struct { + method string + want string + }{ + {method: "getHealth", want: `"ok"`}, + {method: "getVersion", want: `{"feature-set":null,"solana-core":"1.2.3-test"}`}, + {method: "getIdentity", want: `{"identity":"2r1F4iWqVcb8M1DbAjQuFpebkQHY9hcVU4WuW2DJBppN"}`}, + } { + t.Run(test.method, func(t *testing.T) { + request := fmt.Sprintf(`{"jsonrpc":"2.0","id":7,"method":%q}`, test.method) + response, err := endpoint.Client().Post(endpoint.URL, "application/json", strings.NewReader(request)) + require.NoError(t, err) + defer response.Body.Close() + require.Equal(t, http.StatusOK, response.StatusCode) + + var result struct { + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` + Result json.RawMessage `json:"result"` + Error *jsonrpc.JSONRPCError `json:"error"` + } + require.NoError(t, json.NewDecoder(response.Body).Decode(&result)) + require.Equal(t, "2.0", result.JSONRPC) + require.Equal(t, 7, result.ID) + require.Nil(t, result.Error) + require.JSONEq(t, test.want, string(result.Result)) + }) + } +} + +func TestNodeHealthBoundaries(t *testing.T) { + tests := []struct { + name string + ready bool + local uint64 + cluster uint64 + wantHealthy bool + wantBehind *uint64 + }{ + {name: "not ready", cluster: 100}, + {name: "same slot", ready: true, local: 100, cluster: 100, wantHealthy: true}, + {name: "local ahead", ready: true, local: 101, cluster: 100, wantHealthy: true}, + {name: "at threshold", ready: true, local: 100, cluster: 228, wantHealthy: true}, + {name: "past threshold", ready: true, local: 100, cluster: 229, wantBehind: uint64Ptr(129)}, + {name: "large slot without overflow", ready: true, local: ^uint64(0) - 1, cluster: ^uint64(0), wantHealthy: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := nodeHealth(test.ready, test.local, test.cluster) + if test.wantHealthy { + require.NoError(t, err) + require.Equal(t, "ok", got) + return + } + require.Empty(t, got) + var unhealthy *NodeUnhealthyError + require.ErrorAs(t, err, &unhealthy) + require.Equal(t, test.wantBehind, unhealthy.NumSlotsBehind) + }) + } +} + +func TestGetHealthHTTPReportsNotReady(t *testing.T) { + server := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{}, solana.Hash{1}) + require.NoError(t, server.listener.Close()) + endpoint := httptest.NewServer(server) + t.Cleanup(endpoint.Close) + + response, err := endpoint.Client().Post(endpoint.URL, "application/json", strings.NewReader( + `{"jsonrpc":"2.0","id":7,"method":"getHealth"}`, + )) + require.NoError(t, err) + defer response.Body.Close() + + var result struct { + Error *jsonrpc.JSONRPCError `json:"error"` + } + require.NoError(t, json.NewDecoder(response.Body).Decode(&result)) + require.NotNil(t, result.Error) + require.Equal(t, jsonrpc.ErrorCode(-32005), result.Error.Code) + require.Equal(t, "Node is unhealthy", result.Error.Message) + require.Equal(t, map[string]interface{}{"numSlotsBehind": nil}, result.Error.Data) +} + +func TestNodeInfoRejectsParametersAndCancellation(t *testing.T) { + server := &RpcServer{identity: "identity"} + for _, test := range []struct { + name string + call func(context.Context, jsonrpc.RawParams) error + }{ + {name: "getHealth", call: func(ctx context.Context, params jsonrpc.RawParams) error { + _, err := server.GetHealth(ctx, params) + return err + }}, + {name: "getVersion", call: func(ctx context.Context, params jsonrpc.RawParams) error { + _, err := server.GetVersion(ctx, params) + return err + }}, + {name: "getIdentity", call: func(ctx context.Context, params jsonrpc.RawParams) error { + _, err := server.GetIdentity(ctx, params) + return err + }}, + } { + t.Run(test.name, func(t *testing.T) { + var invalid *InvalidParamsError + require.ErrorAs(t, test.call(t.Context(), jsonrpc.RawParams(`[true]`)), &invalid) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + require.ErrorIs(t, test.call(ctx, nil), context.Canceled) + }) + } +} + +func TestGetIdentityRequiresConfiguredValidator(t *testing.T) { + got, err := (&RpcServer{}).GetIdentity(t.Context(), nil) + require.Error(t, err) + require.Empty(t, got.Identity) +} + +func uint64Ptr(value uint64) *uint64 { return &value } From 95b518ae7ebca76a786d4a3173f0c9d39549d210 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Thu, 24 Sep 2026 00:47:13 +0530 Subject: [PATCH 3/6] fix(rpc): base node health on a fresh observed cluster tip --- pkg/blockstream/block_source.go | 36 +++++++++++++++++++++++---------- pkg/replay/block.go | 4 ++++ pkg/rpcserver/node_info.go | 17 ++++++++++++---- pkg/rpcserver/rpcserver.go | 1 + 4 files changed, 43 insertions(+), 15 deletions(-) diff --git a/pkg/blockstream/block_source.go b/pkg/blockstream/block_source.go index 49212382c..e6c6fe384 100644 --- a/pkg/blockstream/block_source.go +++ b/pkg/blockstream/block_source.go @@ -315,15 +315,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 @@ -2465,10 +2467,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. @@ -2515,9 +2526,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 { diff --git a/pkg/replay/block.go b/pkg/replay/block.go index 495ed53a8..c30df9c64 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -2371,6 +2371,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() diff --git a/pkg/rpcserver/node_info.go b/pkg/rpcserver/node_info.go index 23702b3cc..5802cd830 100644 --- a/pkg/rpcserver/node_info.go +++ b/pkg/rpcserver/node_info.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" - "github.com/Overclock-Validator/mithril/pkg/global" "github.com/Overclock-Validator/mithril/pkg/version" "github.com/filecoin-project/go-jsonrpc" ) @@ -27,11 +26,21 @@ func (rpcServer *RpcServer) GetHealth(ctx context.Context, p jsonrpc.RawParams) return "", err } - slotCtx := rpcServer.getSlotCtx() - if slotCtx == nil { + rpcServer.slotCtxMu.RLock() + slotCtx, healthSlot := rpcServer.slotCtx, rpcServer.healthSlot + rpcServer.slotCtxMu.RUnlock() + if slotCtx == nil || healthSlot == nil { return nodeHealth(false, 0, 0) } - return nodeHealth(true, slotCtx.Slot, global.WallClockSlot()) + 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) { diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 8f6f8547b..5ed0002bb 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -30,6 +30,7 @@ type RpcServer struct { epochSchedule *sealevel.SysvarEpochSchedule slotCtx *sealevel.SlotCtx slotCtxMu sync.RWMutex + healthSlot func() (uint64, bool) genesisHash string identity string From 9e6c03c618c83a3591990a499d5ca5ef068435b5 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Thu, 24 Sep 2026 00:47:24 +0530 Subject: [PATCH 4/6] test(rpc): cover missing and stale cluster health observations --- pkg/blockstream/health_slot_test.go | 31 ++++++++++++++++ pkg/rpcserver/node_health_source_test.go | 45 ++++++++++++++++++++++++ pkg/rpcserver/node_info_test.go | 1 + 3 files changed, 77 insertions(+) create mode 100644 pkg/blockstream/health_slot_test.go create mode 100644 pkg/rpcserver/node_health_source_test.go diff --git a/pkg/blockstream/health_slot_test.go b/pkg/blockstream/health_slot_test.go new file mode 100644 index 000000000..6ecec298b --- /dev/null +++ b/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) +} diff --git a/pkg/rpcserver/node_health_source_test.go b/pkg/rpcserver/node_health_source_test.go new file mode 100644 index 000000000..e99cdaa41 --- /dev/null +++ b/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) +} diff --git a/pkg/rpcserver/node_info_test.go b/pkg/rpcserver/node_info_test.go index e79539cee..84aa5bb3d 100644 --- a/pkg/rpcserver/node_info_test.go +++ b/pkg/rpcserver/node_info_test.go @@ -27,6 +27,7 @@ func TestNodeInfoHTTP(t *testing.T) { require.NoError(t, server.listener.Close()) server.SetIdentity("2r1F4iWqVcb8M1DbAjQuFpebkQHY9hcVU4WuW2DJBppN") server.SetSlotCtx(&sealevel.SlotCtx{Slot: global.WallClockSlot()}) + server.SetHealthSlotSource(func() (uint64, bool) { return global.WallClockSlot(), true }) endpoint := httptest.NewServer(server) t.Cleanup(endpoint.Close) From 005cb36e98cfeb3f643d36e797b907c2678e7290 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Wed, 30 Sep 2026 23:40:52 +0530 Subject: [PATCH 5/6] fix(rpc): exclude internal methods from RPC registration --- pkg/rpcserver/rpc_registration_test.go | 89 ++++++++++++++++++++++++++ pkg/rpcserver/rpcserver.go | 22 ++++++- 2 files changed, 110 insertions(+), 1 deletion(-) create mode 100644 pkg/rpcserver/rpc_registration_test.go diff --git a/pkg/rpcserver/rpc_registration_test.go b/pkg/rpcserver/rpc_registration_test.go new file mode 100644 index 000000000..0a8821fde --- /dev/null +++ b/pkg/rpcserver/rpc_registration_test.go @@ -0,0 +1,89 @@ +package rpcserver + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" +) + +func TestRPCHandlerMethodSet(t *testing.T) { + typ := reflect.TypeOf(rpcHandler{}) + if typ.NumMethod() != len(supportedRPCMethods) { + t.Fatalf("registered %d methods, public method list has %d", typ.NumMethod(), len(supportedRPCMethods)) + } + for i := 0; i < typ.NumMethod(); i++ { + name := typ.Method(i).Name + name = strings.ToLower(name[:1]) + name[1:] + if _, ok := supportedRPCMethods[name]; !ok { + t.Fatalf("non-public method registered: %s", name) + } + } +} + +func TestInternalMethodsAreNotRPC(t *testing.T) { + for _, mode := range []string{"single", "mixed-batch", "oversized-single", "direct-dispatch"} { + t.Run(mode, func(t *testing.T) { + rpc := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 32}, solana.Hash{1}) + t.Cleanup(func() { _ = rpc.listener.Close() }) + rpc.SetIdentity("before") + call := map[string]any{"jsonrpc": "2.0", "id": 1, "method": "setIdentity", "params": []any{"after"}} + if mode == "oversized-single" { + call["padding"] = strings.Repeat("x", maxQuietMethodProbeBody+1) + } + var request any = call + if mode == "mixed-batch" { + request = []any{map[string]any{"jsonrpc": "2.0", "id": 2, "method": "getGenesisHash", "params": []any{}}, call} + } + body, err := json.Marshal(request) + if err != nil { + t.Fatal(err) + } + response := httptest.NewRecorder() + if mode == "direct-dispatch" { + rpc.rpcService.HandleRequest(context.Background(), bytes.NewReader(body), response) + } else { + rpc.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))) + } + var replies []struct { + ID int `json:"id"` + Result string `json:"result"` + Error *struct { + Code int `json:"code"` + } `json:"error"` + } + result := response.Body.Bytes() + if mode != "mixed-batch" { + result = append(append([]byte{'['}, result...), ']') + } + if err := json.Unmarshal(result, &replies); err != nil { + t.Fatalf("decode response %q: %v", result, err) + } + found := false + for _, reply := range replies { + if reply.ID == 1 { + found = true + if reply.Error == nil || reply.Error.Code != -32601 { + t.Fatalf("internal method was not rejected: %s", result) + } + } + if reply.ID == 2 && (reply.Error != nil || reply.Result != (solana.Hash{1}).String()) { + t.Fatalf("public batch method failed: %s", result) + } + } + if !found || (mode == "mixed-batch" && len(replies) != 2) { + t.Fatalf("missing response: %s", result) + } + if rpc.identity != "before" { + t.Fatalf("internal request changed identity: %q", rpc.identity) + } + }) + } +} diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 5ed0002bb..3bdec567a 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -21,6 +21,26 @@ import ( "github.com/gagliardetto/solana-go" ) +// rpcHandler exposes only public RPC methods to go-jsonrpc's reflection-based +// registration. Internal setters and lifecycle hooks remain Go-only. +type rpcHandler struct { + rpcMethods +} + +type rpcMethods interface { + GetAccountInfo(context.Context, jsonrpc.RawParams) (GetAccountInfoResp, error) + GetBankHash(context.Context, jsonrpc.RawParams) (string, error) + GetBlockHeight(context.Context, jsonrpc.RawParams) (uint64, error) + GetEpochInfo(context.Context, jsonrpc.RawParams) (GetEpochInfoResp, error) + GetGenesisHash(context.Context, jsonrpc.RawParams) (string, error) + GetHealth(context.Context, jsonrpc.RawParams) (string, error) + GetIdentity(context.Context, jsonrpc.RawParams) (GetIdentityResp, error) + GetLatestBlockhash(context.Context, jsonrpc.RawParams) (GetLatestBlockhashResp, error) + GetVersion(context.Context, jsonrpc.RawParams) (GetVersionResp, error) + SendTransaction(context.Context, jsonrpc.RawParams) (string, error) + SimulateTransaction(context.Context, jsonrpc.RawParams) (SimulateTransactionResp, error) +} + type RpcServer struct { isReady bool rpcService *jsonrpc.RPCServer @@ -82,7 +102,7 @@ func NewRpcServer(acctsDb *accountsdb.AccountsDb, port uint16, epochSchedule *se jsonrpc.WithServerErrors(rpcErrors), ) - rpcServer.rpcService.Register("MithrilRpc", rpcServer) + rpcServer.rpcService.Register("MithrilRpc", rpcHandler{rpcServer}) rpcServer.acctsDb = acctsDb if epochSchedule != nil { rpcServer.epochSchedule = epochSchedule From 54b7b1281921f5938a1358fcfd2b249ba7bfdc27 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Tue, 13 Oct 2026 02:09:12 +0530 Subject: [PATCH 6/6] Fix RPC health reporting during replay recovery --- pkg/replay/alpenglow_switch.go | 6 ++ pkg/replay/alpenglow_unwind_test.go | 14 ++--- pkg/replay/block.go | 16 +++++- pkg/replay/rewards_retirement_test.go | 4 +- pkg/replay/rpc_health_unwind_test.go | 61 +++++++++++++++++++++ pkg/rpcserver/node_health_recovery_test.go | 64 ++++++++++++++++++++++ pkg/rpcserver/node_info.go | 11 +++- pkg/rpcserver/rpc_registration_test.go | 18 +++++- pkg/rpcserver/rpcserver.go | 2 + 9 files changed, 183 insertions(+), 13 deletions(-) create mode 100644 pkg/replay/rpc_health_unwind_test.go create mode 100644 pkg/rpcserver/node_health_recovery_test.go diff --git a/pkg/replay/alpenglow_switch.go b/pkg/replay/alpenglow_switch.go index 6d76cc1fc..7e0ecc63e 100644 --- a/pkg/replay/alpenglow_switch.go +++ b/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 diff --git a/pkg/replay/alpenglow_unwind_test.go b/pkg/replay/alpenglow_unwind_test.go index 52e949c2b..de477d25c 100644 --- a/pkg/replay/alpenglow_unwind_test.go +++ b/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) diff --git a/pkg/replay/block.go b/pkg/replay/block.go index f878ca6a6..87b4ee8c0 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -51,6 +51,10 @@ type SlotCtxSetter interface { SetSlotCtx(slotCtx *sealevel.SlotCtx) } +type rpcHealthInvalidator interface { + InvalidateHealth() +} + // BlockFetchOpts contains options for parallel block fetching type BlockFetchOpts struct { MaxRPS int // Rate limit (requests per second), 0 = use default @@ -1677,6 +1681,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 replayFrontier := uint64(0) if startSlot > 0 { @@ -2481,7 +2490,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]++ @@ -3627,6 +3636,11 @@ func ReplayBlocks( } + // 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()) diff --git a/pkg/replay/rewards_retirement_test.go b/pkg/replay/rewards_retirement_test.go index aaa169ae1..6dc2822bb 100644 --- a/pkg/replay/rewards_retirement_test.go +++ b/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) diff --git a/pkg/replay/rpc_health_unwind_test.go b/pkg/replay/rpc_health_unwind_test.go new file mode 100644 index 000000000..86d0a317f --- /dev/null +++ b/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") + }) + } +} diff --git a/pkg/rpcserver/node_health_recovery_test.go b/pkg/rpcserver/node_health_recovery_test.go new file mode 100644 index 000000000..3fd21933a --- /dev/null +++ b/pkg/rpcserver/node_health_recovery_test.go @@ -0,0 +1,64 @@ +package rpcserver + +import ( + "encoding/json" + "net/http/httptest" + "strings" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/filecoin-project/go-jsonrpc" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestGetHealthHTTPRecoveryPublication(t *testing.T) { + server := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{}, solana.Hash{1}) + require.NoError(t, server.listener.Close()) + endpoint := httptest.NewServer(server) + t.Cleanup(endpoint.Close) + server.SetHealthSlotSource(func() (uint64, bool) { return 1128, true }) + bank := &sealevel.SlotCtx{Slot: 1000} + server.SetSlotCtx(bank) + + check := func(healthy bool, lag *uint64) { + t.Helper() + response, err := endpoint.Client().Post(endpoint.URL, "application/json", strings.NewReader( + `{"jsonrpc":"2.0","id":1,"method":"getHealth"}`, + )) + require.NoError(t, err) + defer response.Body.Close() + var result struct { + Result string `json:"result"` + Error *jsonrpc.JSONRPCError `json:"error"` + } + require.NoError(t, json.NewDecoder(response.Body).Decode(&result)) + if healthy { + require.Nil(t, result.Error) + require.Equal(t, "ok", result.Result) + return + } + require.NotNil(t, result.Error) + require.Equal(t, jsonrpc.ErrorCode(-32005), result.Error.Code) + var expected any + if lag != nil { + expected = float64(*lag) + } + require.Equal(t, map[string]any{"numSlotsBehind": expected}, result.Error.Data) + } + + check(true, nil) + server.InvalidateHealth() + require.Same(t, bank, server.getSlotCtx(), "health invalidation must not clear bank RPC state") + check(false, nil) + // A replacement block source in the checkpoint-recovery loop can observe + // the network before replay publishes any replacement bank. + server.SetHealthSlotSource(func() (uint64, bool) { return 1128, true }) + check(false, nil) + server.SetSlotCtx(&sealevel.SlotCtx{Slot: 999}) + check(false, uint64Ptr(129)) + server.SetSlotCtx(&sealevel.SlotCtx{Slot: 1000}) + check(true, nil) + server.InvalidateHealth() + check(false, nil) +} diff --git a/pkg/rpcserver/node_info.go b/pkg/rpcserver/node_info.go index 5802cd830..49acd6b67 100644 --- a/pkg/rpcserver/node_info.go +++ b/pkg/rpcserver/node_info.go @@ -27,9 +27,9 @@ func (rpcServer *RpcServer) GetHealth(ctx context.Context, p jsonrpc.RawParams) } rpcServer.slotCtxMu.RLock() - slotCtx, healthSlot := rpcServer.slotCtx, rpcServer.healthSlot + slotCtx, healthSlot, ready := rpcServer.slotCtx, rpcServer.healthSlot, rpcServer.healthReady rpcServer.slotCtxMu.RUnlock() - if slotCtx == nil || healthSlot == nil { + if !ready || slotCtx == nil || healthSlot == nil { return nodeHealth(false, 0, 0) } clusterSlot, fresh := healthSlot() @@ -43,6 +43,13 @@ func (rpcServer *RpcServer) SetHealthSlotSource(source func() (uint64, bool)) { rpcServer.slotCtxMu.Unlock() } +// InvalidateHealth marks replay unavailable without discarding the last RPC bank. +func (rpcServer *RpcServer) InvalidateHealth() { + rpcServer.slotCtxMu.Lock() + rpcServer.healthReady = false + rpcServer.slotCtxMu.Unlock() +} + func nodeHealth(ready bool, localSlot, clusterSlot uint64) (string, error) { if !ready { return "", &NodeUnhealthyError{} diff --git a/pkg/rpcserver/rpc_registration_test.go b/pkg/rpcserver/rpc_registration_test.go index 0a8821fde..b7e6396e4 100644 --- a/pkg/rpcserver/rpc_registration_test.go +++ b/pkg/rpcserver/rpc_registration_test.go @@ -29,12 +29,25 @@ func TestRPCHandlerMethodSet(t *testing.T) { } func TestInternalMethodsAreNotRPC(t *testing.T) { + for _, method := range []string{"setIdentity", "invalidateHealth"} { + t.Run(method, func(t *testing.T) { + testInternalMethodNotRPC(t, method) + }) + } +} + +func testInternalMethodNotRPC(t *testing.T, method string) { + t.Helper() for _, mode := range []string{"single", "mixed-batch", "oversized-single", "direct-dispatch"} { t.Run(mode, func(t *testing.T) { rpc := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 32}, solana.Hash{1}) t.Cleanup(func() { _ = rpc.listener.Close() }) rpc.SetIdentity("before") - call := map[string]any{"jsonrpc": "2.0", "id": 1, "method": "setIdentity", "params": []any{"after"}} + rpc.SetSlotCtx(&sealevel.SlotCtx{Slot: 1}) + call := map[string]any{"jsonrpc": "2.0", "id": 1, "method": method, "params": []any{"after"}} + if method == "invalidateHealth" { + call["params"] = []any{} + } if mode == "oversized-single" { call["padding"] = strings.Repeat("x", maxQuietMethodProbeBody+1) } @@ -84,6 +97,9 @@ func TestInternalMethodsAreNotRPC(t *testing.T) { if rpc.identity != "before" { t.Fatalf("internal request changed identity: %q", rpc.identity) } + if !rpc.healthReady { + t.Fatal("internal request invalidated health") + } }) } } diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 3bdec567a..44b37cda7 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -51,6 +51,7 @@ type RpcServer struct { slotCtx *sealevel.SlotCtx slotCtxMu sync.RWMutex healthSlot func() (uint64, bool) + healthReady bool genesisHash string identity string @@ -134,6 +135,7 @@ func fetchAndUnmarshalEpochScheduleSysvar(acctsDb *accountsdb.AccountsDb) *seale func (rpcServer *RpcServer) SetSlotCtx(slotCtx *sealevel.SlotCtx) { rpcServer.slotCtxMu.Lock() rpcServer.slotCtx = slotCtx + rpcServer.healthReady = slotCtx != nil rpcServer.slotCtxMu.Unlock() }