diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index 574e3041..802b2353 100644 --- a/cmd/mithril/node/node.go +++ b/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) } diff --git a/pkg/blockstream/block_source.go b/pkg/blockstream/block_source.go index 32a7fec1..70e43845 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 @@ -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 { 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/alpenglow_switch.go b/pkg/replay/alpenglow_switch.go index 6d76cc1f..7e0ecc63 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 52e949c2..de477d25 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 191f1b73..289026ac 100644 --- a/pkg/replay/block.go +++ b/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()) diff --git a/pkg/replay/rewards_retirement_test.go b/pkg/replay/rewards_retirement_test.go index aaa169ae..6dc2822b 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 00000000..86d0a317 --- /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/errors.go b/pkg/rpcserver/errors.go index 7e8d3893..245cefa2 100644 --- a/pkg/rpcserver/errors.go +++ b/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 } diff --git a/pkg/rpcserver/errors_test.go b/pkg/rpcserver/errors_test.go index 0fa33f45..df3fd51e 100644 --- a/pkg/rpcserver/errors_test.go +++ b/pkg/rpcserver/errors_test.go @@ -98,3 +98,40 @@ func TestEpochRewardHistoryUnavailableError(t *testing.T) { require.Error(t, got.FromJSONRPCError(jsonrpc.JSONRPCError{Code: -32016})) require.Error(t, got.FromJSONRPCError(jsonrpc.JSONRPCError{Code: -32004, Data: map[string]any{"epoch": "invalid"}})) } + +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_recovery_test.go b/pkg/rpcserver/node_health_recovery_test.go new file mode 100644 index 00000000..3fd21933 --- /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_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..49acd6b6 --- /dev/null +++ b/pkg/rpcserver/node_info.go @@ -0,0 +1,93 @@ +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, ready := rpcServer.slotCtx, rpcServer.healthSlot, rpcServer.healthReady + rpcServer.slotCtxMu.RUnlock() + if !ready || 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() +} + +// 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{} + } + 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/rpc_registration_test.go b/pkg/rpcserver/rpc_registration_test.go index 2d781021..be551be8 100644 --- a/pkg/rpcserver/rpc_registration_test.go +++ b/pkg/rpcserver/rpc_registration_test.go @@ -31,12 +31,28 @@ func TestRPCHandlerMethodSet(t *testing.T) { } func TestInternalMethodsAreNotRPC(t *testing.T) { + for _, method := range []string{"enableEpochRewards", "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() }) directory := filepath.Join(t.TempDir(), "history") - call := map[string]any{"jsonrpc": "2.0", "id": 1, "method": "enableEpochRewards", "params": []any{directory, 100, 0}} + rpc.SetIdentity("before") + 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{} + } else if method == "enableEpochRewards" { + call["params"] = []any{directory, 100, 0} + } if mode == "oversized-single" { call["padding"] = strings.Repeat("x", maxQuietMethodProbeBody+1) } @@ -86,6 +102,12 @@ func TestInternalMethodsAreNotRPC(t *testing.T) { if _, err := os.Stat(directory); !os.IsNotExist(err) { t.Fatalf("internal request created history directory: %v", err) } + 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 b8459d61..e3191b4b 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -36,7 +36,10 @@ type rpcMethods interface { GetEpochInfo(context.Context, jsonrpc.RawParams) (GetEpochInfoResp, error) GetGenesisHash(context.Context, jsonrpc.RawParams) (string, error) GetInflationReward(context.Context, jsonrpc.RawParams) ([]*InflationRewardResp, 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) } @@ -53,8 +56,11 @@ type RpcServer struct { epochSchedule *sealevel.SysvarEpochSchedule slotCtx *sealevel.SlotCtx slotCtxMu sync.RWMutex + healthSlot func() (uint64, bool) + healthReady bool genesisHash string epochRewards *epochrewards.Store + identity string leaderTPUCacheMu sync.RWMutex leaderTPUByIdentity map[solana.PublicKey]tpuEndpoint @@ -78,7 +84,10 @@ var supportedRPCMethods = map[string]struct{}{ "getEpochInfo": {}, "getGenesisHash": {}, "getInflationReward": {}, + "getHealth": {}, + "getIdentity": {}, "getLatestBlockhash": {}, + "getVersion": {}, "sendTransaction": {}, "simulateTransaction": {}, } @@ -251,9 +260,14 @@ 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() } +func (rpcServer *RpcServer) SetIdentity(identity string) { + rpcServer.identity = identity +} + func (rpcServer *RpcServer) getSlotCtx() *sealevel.SlotCtx { rpcServer.slotCtxMu.RLock() defer rpcServer.slotCtxMu.RUnlock()