diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index c9c96406..3f018065 100644 --- a/cmd/mithril/node/node.go +++ b/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) } diff --git a/pkg/blockstream/block_source.go b/pkg/blockstream/block_source.go index 6c9b6f4e..ee628757 100644 --- a/pkg/blockstream/block_source.go +++ b/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 { diff --git a/pkg/blockstream/health_slot_test.go b/pkg/blockstream/health_slot_test.go new file mode 100644 index 00000000..6ecec298 --- /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/replay/block.go b/pkg/replay/block.go index e571c98d..bca919c9 100644 --- a/pkg/replay/block.go +++ b/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() diff --git a/pkg/rpcserver/errors.go b/pkg/rpcserver/errors.go index ee2a1654..8ddf704e 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/errors_test.go b/pkg/rpcserver/errors_test.go index 6967eb7d..d6a20365 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_health_source_test.go b/pkg/rpcserver/node_health_source_test.go new file mode 100644 index 00000000..e99cdaa4 --- /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.go b/pkg/rpcserver/node_info.go new file mode 100644 index 00000000..5802cd83 --- /dev/null +++ b/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 +} diff --git a/pkg/rpcserver/node_info_test.go b/pkg/rpcserver/node_info_test.go new file mode 100644 index 00000000..84aa5bb3 --- /dev/null +++ b/pkg/rpcserver/node_info_test.go @@ -0,0 +1,154 @@ +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()}) + server.SetHealthSlotSource(func() (uint64, bool) { return global.WallClockSlot(), true }) + 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 } diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 8a3dd7af..5ed0002b 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -30,7 +30,9 @@ type RpcServer struct { epochSchedule *sealevel.SysvarEpochSchedule slotCtx *sealevel.SlotCtx slotCtxMu sync.RWMutex + healthSlot func() (uint64, bool) genesisHash string + identity string leaderTPUCacheMu sync.RWMutex leaderTPUByIdentity map[solana.PublicKey]tpuEndpoint @@ -53,7 +55,10 @@ var supportedRPCMethods = map[string]struct{}{ "getBlockHeight": {}, "getEpochInfo": {}, "getGenesisHash": {}, + "getHealth": {}, + "getIdentity": {}, "getLatestBlockhash": {}, + "getVersion": {}, "sendTransaction": {}, "simulateTransaction": {}, } @@ -112,6 +117,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()