From 2344cb40a30fd25e025bcb1d5dd61008411e088c Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Wed, 23 Sep 2026 23:30:08 +0530 Subject: [PATCH 1/6] rpc: expose rooted validator metrics for Solana Exporter --- cmd/mithril/node/node.go | 27 ++- pkg/accountsdb/accountsdb.go | 15 ++ pkg/accountsdb/fold.go | 4 +- pkg/replay/block.go | 18 +- pkg/rpcserver/get_account_info.go | 42 +++-- pkg/rpcserver/get_balance.go | 90 ++++++++++ pkg/rpcserver/get_block_production.go | 183 +++++++++++++++++++ pkg/rpcserver/get_epoch_info.go | 60 +++++-- pkg/rpcserver/get_leader_schedule.go | 113 ++++++++++++ pkg/rpcserver/get_vote_accounts.go | 249 ++++++++++++++++++++++++++ pkg/rpcserver/optional_params.go | 10 ++ pkg/rpcserver/parse_vote_account.go | 127 +++++++++++++ pkg/rpcserver/rpcserver.go | 91 ++++++++++ 13 files changed, 1001 insertions(+), 28 deletions(-) create mode 100644 pkg/rpcserver/get_balance.go create mode 100644 pkg/rpcserver/get_block_production.go create mode 100644 pkg/rpcserver/get_leader_schedule.go create mode 100644 pkg/rpcserver/get_vote_accounts.go create mode 100644 pkg/rpcserver/optional_params.go create mode 100644 pkg/rpcserver/parse_vote_account.go diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index c452d494e..337a31bd5 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)) + publishRPCBankState(rpcServer, mithrilState) rpcServer.Start() mlog.Log.Infof("Started RPC server on port %d", rpcPort) } @@ -2877,7 +2878,7 @@ postBootstrap: } } - var slotCtxSetter replay.SlotCtxSetter + var slotCtxSetter replay.RPCStateSetter if rpcServer != nil { slotCtxSetter = rpcServer } @@ -4348,7 +4349,7 @@ func runReplayWithRecovery( useTurbine bool, dbgOpts *replay.DebugOptions, metricsWriter io.Writer, - rpcServer replay.SlotCtxSetter, + rpcServer replay.RPCStateSetter, mithrilState *state.MithrilState, blockFetchOpts *replay.BlockFetchOpts, consensusOpts *replay.ConsensusOpts, @@ -4550,6 +4551,7 @@ func runReplayWithRecovery( mlog.Log.Errorf("fork switch: no rooted checkpoint context to re-replay from; halting") break } + publishRPCBankState(rpcServer, mithrilState) if mithrilState.LastRootedSlot > prevRooted { // progress since last attempt -> fresh budget attempt = 0 prevRooted = mithrilState.LastRootedSlot @@ -4589,3 +4591,24 @@ func runReplayWithRecovery( } return result } + +type rpcBankStateSetter interface { + SetRootedBankState(slot, blockHeight, transactionCount uint64) +} + +func publishRPCBankState(rpcServer rpcBankStateSetter, mithrilState *state.MithrilState) { + if rpcServer == nil || mithrilState == nil { + return + } + rootedSlot := mithrilState.ManifestParentSlot + rootedBlockHeight := mithrilState.ManifestBlockHeight + rootedTransactionCount := mithrilState.ManifestTransactionCount + if rootedCtx := mithrilState.LastRootedContext; rootedCtx != nil { + rootedSlot = rootedCtx.Slot + rootedBlockHeight = rootedCtx.BlockHeight + if rootedCtx.TransactionCount != nil { + rootedTransactionCount = *rootedCtx.TransactionCount + } + } + rpcServer.SetRootedBankState(rootedSlot, rootedBlockHeight, rootedTransactionCount) +} diff --git a/pkg/accountsdb/accountsdb.go b/pkg/accountsdb/accountsdb.go index 49714f731..4c1fd2239 100644 --- a/pkg/accountsdb/accountsdb.go +++ b/pkg/accountsdb/accountsdb.go @@ -97,6 +97,21 @@ func (accountsDb *AccountsDb) StoreQueueLen() int { return accountsDb.inProgressStoreRequests.Len() } +// DurableThrough returns the highest slot whose fold is fully committed. +func (accountsDb *AccountsDb) DurableThrough() uint64 { + return accountsDb.durableThrough.Load() +} + +// VoteAccountPubkeys returns the vote accounts in the current cache view. +func (accountsDb *AccountsDb) VoteAccountPubkeys() []solana.PublicKey { + pubkeys := make([]solana.PublicKey, 0, accountsDb.VoteAcctCache.Size()) + accountsDb.VoteAcctCache.Range(func(pubkey solana.PublicKey, _ *accounts.Account) bool { + pubkeys = append(pubkeys, pubkey) + return true + }) + return pubkeys +} + // silentLogger implements pebble.Logger but discards all messages. // This suppresses verbose WAL recovery messages on startup. type silentLogger struct{} diff --git a/pkg/accountsdb/fold.go b/pkg/accountsdb/fold.go index 096a59a16..7e01dfe8d 100644 --- a/pkg/accountsdb/fold.go +++ b/pkg/accountsdb/fold.go @@ -304,6 +304,9 @@ func (db *AccountsDb) CommitBatch( // ordinary concurrent-safe cache operations finish. db.refreshReadCacheEntries(live) + // Advance the watermark before removing pendingFold so rooted readers keep + // waiting until the matching bank state is published. + db.durableThrough.Store(throughSlot) db.readCacheEpochMu.Lock() db.pendingFold = nil db.readCacheEpochMu.Unlock() @@ -311,7 +314,6 @@ func (db *AccountsDb) CommitBatch( // (8) Publish. db.lastBatchSeq = batchSeq - db.durableThrough.Store(throughSlot) return BatchCommitResult{ BatchSeq: batchSeq, diff --git a/pkg/replay/block.go b/pkg/replay/block.go index 495ed53a8..8af6b03fd 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -49,9 +49,11 @@ import ( "github.com/panjf2000/ants/v2" ) -// SlotCtxSetter is implemented by types that accept a SlotCtx update (e.g. RpcServer). -type SlotCtxSetter interface { +// RPCStateSetter is implemented by the RPC server so replay can publish both +// its live execution bank and the latest bank durably folded into AccountsDB. +type RPCStateSetter interface { SetSlotCtx(slotCtx *sealevel.SlotCtx) + SetRootedBankState(slot, blockHeight, transactionCount uint64) } // BlockFetchOpts contains options for parallel block fetching @@ -1670,7 +1672,7 @@ func ReplayBlocks( useTurbine bool, dbgOpts *DebugOptions, metricsWriter io.Writer, - rpcServer SlotCtxSetter, + rpcServer RPCStateSetter, blockFetchOpts *BlockFetchOpts, consensusOpts *ConsensusOpts, // nil = use defaults (max_depth=64, policy="halt") onCancelWriteState OnCancelWriteState, // callback to write state immediately on cancellation (can be nil) @@ -2063,6 +2065,13 @@ func ReplayBlocks( mithrilState.LastRootedSlot = promotedThrough mithrilState.LastRootedBankhash = rootedCtx.Bankhash mithrilState.LastRootedContext = rootedCtx + if rpcServer != nil { + var transactionCount uint64 + if rootedCtx.TransactionCount != nil { + transactionCount = *rootedCtx.TransactionCount + } + rpcServer.SetRootedBankState(promotedThrough, rootedCtx.BlockHeight, transactionCount) + } if transactionStatuses.Root(promotedThrough) { mlog.Log.Infof("transaction status cache reconstructed complete %d-root coverage through durable slot %d", maxTransactionStatusRoots, promotedThrough) @@ -3085,6 +3094,9 @@ func ReplayBlocks( if rpcServer != nil { rpcServer.SetSlotCtx(lastSlotCtx) + if unrootedTailState == nil { + rpcServer.SetRootedBankState(block.Slot, global.BlockHeight(), global.TransactionCount()) + } } replayCtx.Capitalization -= lastSlotCtx.LamportsBurnt diff --git a/pkg/rpcserver/get_account_info.go b/pkg/rpcserver/get_account_info.go index 11d01e4c2..3909505e0 100644 --- a/pkg/rpcserver/get_account_info.go +++ b/pkg/rpcserver/get_account_info.go @@ -3,11 +3,13 @@ package rpcserver import ( "context" "encoding/base64" + "errors" "fmt" "reflect" "github.com/DataDog/zstd" - "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/accountsdb" + "github.com/Overclock-Validator/mithril/pkg/addresses" "github.com/Overclock-Validator/mithril/pkg/mlog" "github.com/Overclock-Validator/mithril/pkg/safemath" "github.com/filecoin-project/go-jsonrpc" @@ -73,18 +75,31 @@ func (rpcServer *RpcServer) GetAccountInfo(ctx context.Context, p jsonrpc.RawPar return GetAccountInfoResp{}, fmt.Errorf("invalid base58 encoding") } - acct, err := rpcServer.acctsDb.GetAccount(0, pk) - if err != nil { - return GetAccountInfoResp{}, nil - } - conf, err := parseGetAccountInfoConfMap(params) if err != nil { return GetAccountInfoResp{}, err } + rooted, acct, err := rpcServer.readRootedAccount(ctx, pk) + if err != nil && !errors.Is(err, accountsdb.ErrNoAccount) { + return GetAccountInfoResp{}, fmt.Errorf("read account at rooted slot %d: %w", rooted.Slot, err) + } + if conf.MinContextSlot > rooted.Slot { + return GetAccountInfoResp{}, &MinContextSlotNotReachedError{ContextSlot: rooted.Slot} + } + if errors.Is(err, accountsdb.ErrNoAccount) { + return GetAccountInfoResp{Context: GetAccountInfoRespContext{ApiVersion: "mithril 0.1", Slot: rooted.Slot}}, nil + } var acctData interface{} - acctData, err = encodeAcctDataWithConfig(acct.Data, conf) + if conf.EncodingType != nil && *conf.EncodingType == GetAccountEncodingJson && conf.DataSlice != nil { + return GetAccountInfoResp{}, fmt.Errorf("cannot use jsonParsed with dataSlice") + } + if conf.EncodingType != nil && *conf.EncodingType == GetAccountEncodingJson && acct.Owner == addresses.VoteProgramAddr { + acctData, err = parsedVoteAccountData(acct.Data, pk) + } + if acctData == nil || err != nil { + acctData, err = encodeAcctDataWithConfig(acct.Data, conf) + } if err != nil { return GetAccountInfoResp{}, err } @@ -97,7 +112,7 @@ func (rpcServer *RpcServer) GetAccountInfo(ctx context.Context, p jsonrpc.RawPar RentEpoch: acct.RentEpoch, Space: uint64(len(acct.Data))} - return GetAccountInfoResp{Context: GetAccountInfoRespContext{ApiVersion: "mithril 0.1", Slot: global.Slot()}, Value: val}, nil + return GetAccountInfoResp{Context: GetAccountInfoRespContext{ApiVersion: "mithril 0.1", Slot: rooted.Slot}, Value: val}, nil } func parseGetAccountInfoConfMap(params []interface{}) (*GetAccountInfoConfig, error) { @@ -117,12 +132,19 @@ func parseGetAccountInfoConfMap(params []interface{}) (*GetAccountInfoConfig, er commitmentObj, ok := confMap["commitment"] if ok { commitmentStr, ok := commitmentObj.(string) - if !ok { + if !ok || (commitmentStr != "processed" && commitmentStr != "confirmed" && commitmentStr != "finalized") { return nil, fmt.Errorf("invalid commitment") } conf.Commitment = commitmentStr } + if minContextSlot, ok := confMap["minContextSlot"]; ok { + conf.MinContextSlot, err = rpcUint64(minContextSlot, "minContextSlot") + if err != nil { + return nil, err + } + } + // encoding type is optional. defaults to base58. encodingObj, ok := confMap["encoding"] if ok { @@ -199,7 +221,7 @@ func parseGetAcctDataEncodingType(encodingStr string) (int, error) { return GetAccountEncodingBase64Zstd, nil case "jsonParsed": - return GetAccountEncodingBase64, nil + return GetAccountEncodingJson, nil default: return 0, fmt.Errorf("invalid data encoding %s", encodingStr) diff --git a/pkg/rpcserver/get_balance.go b/pkg/rpcserver/get_balance.go new file mode 100644 index 000000000..dcc9036b0 --- /dev/null +++ b/pkg/rpcserver/get_balance.go @@ -0,0 +1,90 @@ +package rpcserver + +import ( + "context" + "errors" + "fmt" + + "github.com/Overclock-Validator/mithril/pkg/accountsdb" + "github.com/filecoin-project/go-jsonrpc" + "github.com/gagliardetto/solana-go" +) + +type GetBalanceResp struct { + Context GetBalanceRespContext `json:"context"` + Value uint64 `json:"value"` +} + +type GetBalanceRespContext struct { + APIVersion string `json:"apiVersion"` + Slot uint64 `json:"slot"` +} + +func (rpcServer *RpcServer) GetBalance(ctx context.Context, p jsonrpc.RawParams) (GetBalanceResp, error) { + if err := ctx.Err(); err != nil { + return GetBalanceResp{}, err + } + + params, err := jsonrpc.DecodeParams[[]interface{}](p) + if err != nil { + return GetBalanceResp{}, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + if len(params) < 1 || len(params) > 2 { + return GetBalanceResp{}, &InvalidParamsError{Message: "getBalance requires an address and optional config"} + } + address, ok := params[0].(string) + if !ok { + return GetBalanceResp{}, &InvalidParamsError{Message: "getBalance requires an address as first parameter"} + } + pubkey, err := solana.PublicKeyFromBase58(address) + if err != nil { + return GetBalanceResp{}, &InvalidParamsError{Message: "Invalid param: Invalid"} + } + + minContextSlot, err := parseBalanceConfig(params) + if err != nil { + return GetBalanceResp{}, err + } + rooted, account, err := rpcServer.readRootedAccount(ctx, pubkey) + if err != nil && !errors.Is(err, accountsdb.ErrNoAccount) { + return GetBalanceResp{}, fmt.Errorf("read balance at rooted slot %d: %w", rooted.Slot, err) + } + if minContextSlot != nil && rooted.Slot < *minContextSlot { + return GetBalanceResp{}, &MinContextSlotNotReachedError{ContextSlot: rooted.Slot} + } + + var lamports uint64 + if account != nil { + lamports = account.Lamports + } + + return GetBalanceResp{ + Context: GetBalanceRespContext{APIVersion: "mithril 0.1", Slot: rooted.Slot}, + Value: lamports, + }, nil +} + +func parseBalanceConfig(params []interface{}) (*uint64, error) { + if len(params) == 1 || params[1] == nil { + return nil, nil + } + config, ok := params[1].(map[string]interface{}) + if !ok { + return nil, &InvalidParamsError{Message: "invalid getBalance config"} + } + if commitment, exists := config["commitment"]; exists { + value, ok := commitment.(string) + if !ok || (value != "processed" && value != "confirmed" && value != "finalized") { + return nil, &InvalidParamsError{Message: "invalid commitment"} + } + } + value, exists := config["minContextSlot"] + if !exists { + return nil, nil + } + minContextSlot, err := rpcUint64(value, "minContextSlot") + if err != nil { + return nil, err + } + return &minContextSlot, nil +} diff --git a/pkg/rpcserver/get_block_production.go b/pkg/rpcserver/get_block_production.go new file mode 100644 index 000000000..54385ccf1 --- /dev/null +++ b/pkg/rpcserver/get_block_production.go @@ -0,0 +1,183 @@ +package rpcserver + +import ( + "context" + "fmt" + "math" + + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/filecoin-project/go-jsonrpc" + bin "github.com/gagliardetto/binary" + "github.com/gagliardetto/solana-go" +) + +type GetBlockProductionResp struct { + Context GetBalanceRespContext `json:"context"` + Value BlockProductionResults `json:"value"` +} + +type BlockProductionResults struct { + ByIdentity map[string][2]uint64 `json:"byIdentity"` + Range BlockProductionRange `json:"range"` +} + +type BlockProductionRange struct { + FirstSlot uint64 `json:"firstSlot"` + LastSlot uint64 `json:"lastSlot"` +} + +type getBlockProductionConfig struct { + identity *solana.PublicKey + firstSlot *uint64 + lastSlot *uint64 +} + +func (rpcServer *RpcServer) GetBlockProduction(ctx context.Context, p jsonrpc.RawParams) (GetBlockProductionResp, error) { + if err := ctx.Err(); err != nil { + return GetBlockProductionResp{}, err + } + params, err := decodeOptionalParams(p) + if err != nil { + return GetBlockProductionResp{}, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + config, err := parseBlockProductionConfig(params) + if err != nil { + return GetBlockProductionResp{}, err + } + if rpcServer.epochSchedule == nil { + return GetBlockProductionResp{}, fmt.Errorf("node has no epoch schedule available") + } + rooted, historyAccount, err := rpcServer.readRootedAccount(ctx, sealevel.SysvarSlotHistoryAddr) + if err != nil { + return GetBlockProductionResp{}, fmt.Errorf("read slot history at rooted slot %d: %w", rooted.Slot, err) + } + + firstSlot := rpcServer.epochSchedule.FirstSlotInEpoch(rpcServer.epochSchedule.GetEpoch(rooted.Slot)) + if config.firstSlot != nil { + firstSlot = *config.firstSlot + } + lastSlot := rooted.Slot + if config.lastSlot != nil { + lastSlot = *config.lastSlot + } + if lastSlot < firstSlot { + return GetBlockProductionResp{}, &InvalidParamsError{Message: fmt.Sprintf("lastSlot, %d, cannot be less than firstSlot, %d", lastSlot, firstSlot)} + } + + var history sealevel.SysvarSlotHistory + if err := history.UnmarshalWithDecoder(bin.NewBinDecoder(historyAccount.Data)); err != nil { + return GetBlockProductionResp{}, fmt.Errorf("decode slot history at rooted slot %d: %w", rooted.Slot, err) + } + oldest, newest, err := slotHistoryBounds(&history) + if err != nil { + return GetBlockProductionResp{}, err + } + if firstSlot < oldest { + return GetBlockProductionResp{}, &InvalidParamsError{Message: fmt.Sprintf("firstSlot, %d, is too small; min %d", firstSlot, oldest)} + } + if lastSlot > newest { + return GetBlockProductionResp{}, &InvalidParamsError{Message: fmt.Sprintf("lastSlot, %d, is too large; max %d", lastSlot, newest)} + } + + byIdentity := make(map[string][2]uint64) + for slot := firstSlot; ; slot++ { + leader, ok := global.LeaderForSlot(slot) + if !ok { + return GetBlockProductionResp{}, fmt.Errorf("leader schedule unavailable for slot %d", slot) + } + if config.identity == nil || leader == *config.identity { + production := byIdentity[leader.String()] + production[0]++ + if slotHistoryContains(&history, slot) { + production[1]++ + } + byIdentity[leader.String()] = production + } + if slot == lastSlot { + break + } + } + + return GetBlockProductionResp{ + Context: GetBalanceRespContext{APIVersion: "mithril 0.1", Slot: rooted.Slot}, + Value: BlockProductionResults{ + ByIdentity: byIdentity, + Range: BlockProductionRange{FirstSlot: firstSlot, LastSlot: lastSlot}, + }, + }, nil +} + +func parseBlockProductionConfig(params []interface{}) (getBlockProductionConfig, error) { + var config getBlockProductionConfig + if len(params) > 1 { + return config, &InvalidParamsError{Message: "getBlockProduction accepts at most one config object"} + } + if len(params) == 0 || params[0] == nil { + return config, nil + } + values, ok := params[0].(map[string]interface{}) + if !ok { + return config, &InvalidParamsError{Message: "invalid getBlockProduction config"} + } + if commitment, exists := values["commitment"]; exists { + value, ok := commitment.(string) + if !ok || (value != "processed" && value != "confirmed" && value != "finalized") { + return config, &InvalidParamsError{Message: "invalid commitment"} + } + } + if raw, exists := values["identity"]; exists { + value, ok := raw.(string) + if !ok { + return config, &InvalidParamsError{Message: "invalid identity"} + } + identity, err := solana.PublicKeyFromBase58(value) + if err != nil { + return config, &InvalidParamsError{Message: "invalid identity"} + } + config.identity = &identity + } + if raw, exists := values["range"]; exists { + rangeValues, ok := raw.(map[string]interface{}) + if !ok { + return config, &InvalidParamsError{Message: "invalid range"} + } + first, exists := rangeValues["firstSlot"] + if !exists { + return config, &InvalidParamsError{Message: "range requires firstSlot"} + } + firstSlot, err := rpcUint64(first, "firstSlot") + if err != nil { + return config, err + } + config.firstSlot = &firstSlot + if last, exists := rangeValues["lastSlot"]; exists && last != nil { + lastSlot, err := rpcUint64(last, "lastSlot") + if err != nil { + return config, err + } + config.lastSlot = &lastSlot + } + } + return config, nil +} + +func rpcUint64(raw interface{}, name string) (uint64, error) { + value, ok := raw.(float64) + if !ok || value < 0 || value >= math.Exp2(64) || math.Trunc(value) != value { + return 0, &InvalidParamsError{Message: fmt.Sprintf("invalid %s", name)} + } + return uint64(value), nil +} + +func slotHistoryBounds(history *sealevel.SysvarSlotHistory) (uint64, uint64, error) { + if history == nil || history.Bits.Len == 0 || history.Bits.Bits.BlocksLen == 0 || len(history.Bits.Bits.Blocks) < int(history.Bits.Bits.BlocksLen) || history.NextSlot == 0 { + return 0, 0, fmt.Errorf("node has no usable slot history") + } + return history.NextSlot - min(history.NextSlot, history.Bits.Len), history.NextSlot - 1, nil +} + +func slotHistoryContains(history *sealevel.SysvarSlotHistory, slot uint64) bool { + block := (slot / 64) % history.Bits.Bits.BlocksLen + return history.Bits.Bits.Blocks[block]&(uint64(1)<<(slot%64)) != 0 +} diff --git a/pkg/rpcserver/get_epoch_info.go b/pkg/rpcserver/get_epoch_info.go index 2ee4a9cdc..5f4ee494e 100644 --- a/pkg/rpcserver/get_epoch_info.go +++ b/pkg/rpcserver/get_epoch_info.go @@ -18,25 +18,61 @@ type GetEpochInfoResp struct { } func (rpcServer *RpcServer) GetEpochInfo(ctx context.Context, p jsonrpc.RawParams) (GetEpochInfoResp, error) { - params, err := jsonrpc.DecodeParams[[]interface{}](p) + params, err := decodeOptionalParams(p) if err != nil { - return GetEpochInfoResp{}, fmt.Errorf("decoding params: %w", err) + return GetEpochInfoResp{}, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + if len(params) > 1 { + return GetEpochInfoResp{}, &InvalidParamsError{Message: "getEpochInfo accepts at most one config object"} + } + commitment := "finalized" + var minContextSlot *uint64 + if len(params) == 1 && params[0] != nil { + config, ok := params[0].(map[string]interface{}) + if !ok { + return GetEpochInfoResp{}, &InvalidParamsError{Message: "invalid getEpochInfo config"} + } + if raw, exists := config["commitment"]; exists { + value, ok := raw.(string) + if !ok || (value != "processed" && value != "confirmed" && value != "finalized") { + return GetEpochInfoResp{}, &InvalidParamsError{Message: "invalid commitment"} + } + commitment = value + } + if raw, exists := config["minContextSlot"]; exists { + value, err := rpcUint64(raw, "minContextSlot") + if err != nil { + return GetEpochInfoResp{}, err + } + minContextSlot = &value + } + } + if rpcServer.epochSchedule == nil { + return GetEpochInfoResp{}, fmt.Errorf("node has no epoch schedule available") } - - _ = params - - epoch := global.Epoch() slot := global.Slot() - firstSlotInEpoch := rpcServer.epochSchedule.FirstSlotInEpoch(epoch) - slotIndex := slot - firstSlotInEpoch + blockHeight := global.BlockHeight() + transactionCount := global.TransactionCount() + if commitment != "processed" { + rooted, ok := rpcServer.getRootedBankState() + if !ok { + return GetEpochInfoResp{}, fmt.Errorf("node has no rooted bank available") + } + slot, blockHeight, transactionCount = rooted.Slot, rooted.BlockHeight, rooted.TransactionCount + } + if minContextSlot != nil && slot < *minContextSlot { + return GetEpochInfoResp{}, &MinContextSlotNotReachedError{ContextSlot: slot} + } + epoch := rpcServer.epochSchedule.GetEpoch(slot) + slotIndex := slot - rpcServer.epochSchedule.FirstSlotInEpoch(epoch) resp := GetEpochInfoResp{ - AbsoluteSlot: global.Slot(), - BlockHeight: global.BlockHeight(), - Epoch: global.Epoch(), + AbsoluteSlot: slot, + BlockHeight: blockHeight, + Epoch: epoch, SlotIndex: slotIndex, SlotsInEpoch: 432000, - TransactionCount: global.TransactionCount(), + TransactionCount: transactionCount, } return resp, nil diff --git a/pkg/rpcserver/get_leader_schedule.go b/pkg/rpcserver/get_leader_schedule.go new file mode 100644 index 000000000..0310d3f2e --- /dev/null +++ b/pkg/rpcserver/get_leader_schedule.go @@ -0,0 +1,113 @@ +package rpcserver + +import ( + "context" + "fmt" + + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/filecoin-project/go-jsonrpc" + "github.com/gagliardetto/solana-go" +) + +type getLeaderScheduleConfig struct { + identity *solana.PublicKey +} + +func (rpcServer *RpcServer) GetLeaderSchedule(ctx context.Context, p jsonrpc.RawParams) (map[string][]uint64, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + params, err := decodeOptionalParams(p) + if err != nil { + return nil, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + if len(params) > 2 { + return nil, &InvalidParamsError{Message: "getLeaderSchedule accepts an optional slot and config"} + } + rooted, ok := rpcServer.getRootedBankState() + if !ok { + return nil, fmt.Errorf("node has no rooted bank available") + } + if rpcServer.epochSchedule == nil { + return nil, fmt.Errorf("node has no epoch schedule available") + } + + slot := rooted.Slot + var rawConfig interface{} + if len(params) > 0 && params[0] != nil { + if _, ok := params[0].(map[string]interface{}); ok { + if len(params) != 1 { + return nil, &InvalidParamsError{Message: "leader schedule config cannot be provided twice"} + } + rawConfig = params[0] + } else { + value, err := rpcUint64(params[0], "slot") + if err != nil { + return nil, err + } + slot = value + } + } + if len(params) == 2 { + rawConfig = params[1] + } + config, err := parseLeaderScheduleConfig(rawConfig) + if err != nil { + return nil, err + } + + epoch := rpcServer.epochSchedule.GetEpoch(slot) + firstSlot := rpcServer.epochSchedule.FirstSlotInEpoch(epoch) + slotsInEpoch := rpcServer.epochSchedule.SlotsInEpoch(epoch) + schedule := make(map[string][]uint64) + for index := uint64(0); index < slotsInEpoch; index++ { + leader, ok := global.LeaderForSlot(firstSlot + index) + if !ok { + return nil, nil + } + if config.identity != nil && leader != *config.identity { + continue + } + key := leader.String() + schedule[key] = append(schedule[key], index) + } + return schedule, nil +} + +func parseLeaderScheduleConfig(raw interface{}) (getLeaderScheduleConfig, error) { + var config getLeaderScheduleConfig + if raw == nil { + return config, nil + } + values, ok := raw.(map[string]interface{}) + if !ok { + return config, &InvalidParamsError{Message: "invalid getLeaderSchedule config"} + } + if commitment, exists := values["commitment"]; exists { + value, ok := commitment.(string) + if !ok || (value != "processed" && value != "confirmed" && value != "finalized") { + return config, &InvalidParamsError{Message: "invalid commitment"} + } + } + if raw, exists := values["identity"]; exists { + value, ok := raw.(string) + if !ok { + return config, &InvalidParamsError{Message: "invalid identity"} + } + identity, err := solana.PublicKeyFromBase58(value) + if err != nil { + return config, &InvalidParamsError{Message: "invalid identity"} + } + config.identity = &identity + } + if raw, exists := values["keyByVoteAccount"]; exists { + value, ok := raw.(bool) + if !ok { + return config, &InvalidParamsError{Message: "invalid keyByVoteAccount"} + } + if value { + return config, &InvalidParamsError{Message: "keyByVoteAccount is not supported"} + } + } + return config, nil +} diff --git a/pkg/rpcserver/get_vote_accounts.go b/pkg/rpcserver/get_vote_accounts.go new file mode 100644 index 000000000..b6eddddfe --- /dev/null +++ b/pkg/rpcserver/get_vote_accounts.go @@ -0,0 +1,249 @@ +package rpcserver + +import ( + "context" + "fmt" + "slices" + "sync" + + "github.com/Overclock-Validator/mithril/pkg/addresses" + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/filecoin-project/go-jsonrpc" + bin "github.com/gagliardetto/binary" + "github.com/gagliardetto/solana-go" +) + +const ( + delinquentValidatorSlotDistance = 128 + maxRPCEpochCreditsHistory = 5 +) + +type GetVoteAccountsResp struct { + Current []VoteAccountInfo `json:"current"` + Delinquent []VoteAccountInfo `json:"delinquent"` +} + +type VoteAccountInfo struct { + VotePubkey string `json:"votePubkey"` + NodePubkey string `json:"nodePubkey"` + ActivatedStake uint64 `json:"activatedStake"` + Commission uint8 `json:"commission"` + InflationRewardsCommissionBPS uint16 `json:"inflationRewardsCommissionBps"` + EpochCredits [][3]uint64 `json:"epochCredits"` + EpochVoteAccount bool `json:"epochVoteAccount"` + LastVote uint64 `json:"lastVote"` + RootSlot uint64 `json:"rootSlot"` +} + +type getVoteAccountsConfig struct { + votePubkey *solana.PublicKey + delinquentSlotDistance uint64 +} + +func (rpcServer *RpcServer) GetVoteAccounts(ctx context.Context, p jsonrpc.RawParams) (GetVoteAccountsResp, error) { + if err := ctx.Err(); err != nil { + return GetVoteAccountsResp{}, err + } + params, err := decodeOptionalParams(p) + if err != nil { + return GetVoteAccountsResp{}, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + config, err := parseGetVoteAccountsConfig(params) + if err != nil { + return GetVoteAccountsResp{}, err + } + if rpcServer.epochSchedule == nil { + return GetVoteAccountsResp{}, fmt.Errorf("node has no epoch schedule available") + } + if rpcServer.acctsDb == nil { + return GetVoteAccountsResp{}, fmt.Errorf("node has no accounts database available") + } + + for { + rooted, ok := rpcServer.getRootedBankState() + if !ok { + return GetVoteAccountsResp{}, fmt.Errorf("node has no rooted bank available") + } + epoch := rpcServer.epochSchedule.GetEpoch(rooted.Slot) + stakes, ok := global.EpochStakesSnapshot(epoch) + if !ok { + return GetVoteAccountsResp{}, fmt.Errorf("epoch stakes unavailable for rooted epoch %d", epoch) + } + activeStakes, err := rpcServer.rootedActivatedStakes(ctx, rooted.Slot, epoch) + if err != nil { + return GetVoteAccountsResp{}, err + } + + votePubkeys := rpcServer.acctsDb.VoteAccountPubkeys() + // A fresh snapshot can have an empty vote cache before any vote account is read. + for votePubkey := range stakes.Stakes { + votePubkeys = append(votePubkeys, votePubkey) + } + slices.SortFunc(votePubkeys, func(left, right solana.PublicKey) int { + return slices.Compare(left[:], right[:]) + }) + votePubkeys = slices.Compact(votePubkeys) + filtered := votePubkeys[:0] + for _, votePubkey := range votePubkeys { + if config.votePubkey == nil || votePubkey == *config.votePubkey { + filtered = append(filtered, votePubkey) + } + } + votePubkeys = filtered + published, accounts, err := rpcServer.readRootedAccounts(ctx, votePubkeys) + if err != nil { + return GetVoteAccountsResp{}, fmt.Errorf("read vote accounts at rooted slot %d: %w", rooted.Slot, err) + } + if published.Slot != rooted.Slot { + continue + } + + response := GetVoteAccountsResp{ + Current: make([]VoteAccountInfo, 0, len(votePubkeys)), + Delinquent: make([]VoteAccountInfo, 0), + } + for index, votePubkey := range votePubkeys { + account := accounts[index] + if account == nil || account.Owner != addresses.VoteProgramAddr { + return GetVoteAccountsResp{}, fmt.Errorf("vote account %s is unavailable at rooted slot %d", votePubkey, rooted.Slot) + } + versioned, err := sealevel.UnmarshalVersionedVoteState(account.Data) + if err != nil { + return GetVoteAccountsResp{}, fmt.Errorf("decode vote account %s: %w", votePubkey, err) + } + voteState := versioned.ConvertToCurrent() + lastVote, _ := voteState.LastVotedSlot() + var rootSlot uint64 + if voteState.RootSlot != nil { + rootSlot = *voteState.RootSlot + } + creditsStart := max(0, len(voteState.EpochCredits)-maxRPCEpochCreditsHistory) + creditsHistory := voteState.EpochCredits[creditsStart:] + epochCredits := make([][3]uint64, len(creditsHistory)) + for i, credits := range creditsHistory { + epochCredits[i] = [3]uint64{credits.Epoch, credits.Credits, credits.PrevCredits} + } + commission, commissionBPS := voteCommission(versioned, voteState) + _, epochVoteAccount := stakes.Stakes[votePubkey] + info := VoteAccountInfo{ + VotePubkey: votePubkey.String(), + NodePubkey: voteState.NodePubkey.String(), + ActivatedStake: activeStakes[votePubkey], + Commission: commission, + InflationRewardsCommissionBPS: commissionBPS, + EpochCredits: epochCredits, + EpochVoteAccount: epochVoteAccount, + LastVote: lastVote, + RootSlot: rootSlot, + } + current := lastVote > 0 + if rooted.Slot >= config.delinquentSlotDistance { + current = lastVote > rooted.Slot-config.delinquentSlotDistance + } + if current { + response.Current = append(response.Current, info) + } else if info.ActivatedStake > 0 { + response.Delinquent = append(response.Delinquent, info) + } + } + return response, nil + } +} + +func (rpcServer *RpcServer) rootedActivatedStakes(ctx context.Context, slot, epoch uint64) (map[solana.PublicKey]uint64, error) { + rooted, accounts, err := rpcServer.readRootedAccounts(ctx, []solana.PublicKey{sealevel.SysvarStakeHistoryAddr}) + if err != nil { + return nil, fmt.Errorf("read rooted stake history: %w", err) + } + if rooted.Slot != slot { + return nil, nil + } + if accounts[0] == nil { + return nil, fmt.Errorf("stake history unavailable at rooted slot %d", slot) + } + var history sealevel.SysvarStakeHistory + if err := history.UnmarshalWithDecoder(bin.NewBinDecoder(accounts[0].Data)); err != nil { + return nil, fmt.Errorf("decode stake history at rooted slot %d: %w", slot, err) + } + slotCtx := rpcServer.getSlotCtx() + if slotCtx == nil || slotCtx.Features == nil { + return nil, fmt.Errorf("node features unavailable for stake activation") + } + activationEpoch := sealevel.NewWarmupCooldownRateEpochWithSlotCtx(slotCtx, rpcServer.epochSchedule) + stakes := make(map[solana.PublicKey]uint64) + var mu sync.Mutex + _, err = global.StreamStakeAccounts(rpcServer.acctsDb, slot, func(_ solana.PublicKey, delegation *sealevel.Delegation, _ uint64) { + active := delegation.Stake(epoch, &history, activationEpoch) + if active == 0 { + return + } + mu.Lock() + stakes[delegation.VoterPubkey] += active + mu.Unlock() + }) + if err != nil { + return nil, fmt.Errorf("scan rooted stake accounts at slot %d: %w", slot, err) + } + if err := ctx.Err(); err != nil { + return nil, err + } + return stakes, nil +} + +func parseGetVoteAccountsConfig(params []interface{}) (getVoteAccountsConfig, error) { + config := getVoteAccountsConfig{delinquentSlotDistance: delinquentValidatorSlotDistance} + if len(params) > 1 { + return config, &InvalidParamsError{Message: "getVoteAccounts accepts at most one config object"} + } + if len(params) == 0 || params[0] == nil { + return config, nil + } + values, ok := params[0].(map[string]interface{}) + if !ok { + return config, &InvalidParamsError{Message: "invalid getVoteAccounts config"} + } + if commitment, exists := values["commitment"]; exists { + value, ok := commitment.(string) + if !ok || (value != "processed" && value != "confirmed" && value != "finalized") { + return config, &InvalidParamsError{Message: "invalid commitment"} + } + } + if raw, exists := values["votePubkey"]; exists { + value, ok := raw.(string) + if !ok { + return config, &InvalidParamsError{Message: "invalid votePubkey"} + } + pubkey, err := solana.PublicKeyFromBase58(value) + if err != nil { + return config, &InvalidParamsError{Message: "invalid votePubkey"} + } + config.votePubkey = &pubkey + } + if raw, exists := values["keepUnstakedDelinquents"]; exists { + value, ok := raw.(bool) + if !ok { + return config, &InvalidParamsError{Message: "invalid keepUnstakedDelinquents"} + } + if value { + return config, &InvalidParamsError{Message: "keepUnstakedDelinquents is not supported"} + } + } + if raw, exists := values["delinquentSlotDistance"]; exists { + value, err := rpcUint64(raw, "delinquentSlotDistance") + if err != nil { + return config, err + } + config.delinquentSlotDistance = value + } + return config, nil +} + +func voteCommission(versioned *sealevel.VoteStateVersions, current *sealevel.VoteState) (uint8, uint16) { + if versioned.Type == sealevel.VoteStateVersionV4 { + basisPoints := versioned.V4.InflationRewardsCommissionBps + percent := (uint32(basisPoints) + 99) / 100 + return uint8(min(uint32(255), percent)), basisPoints + } + return current.Commission, uint16(current.Commission) * 100 +} diff --git a/pkg/rpcserver/optional_params.go b/pkg/rpcserver/optional_params.go new file mode 100644 index 000000000..74c000e25 --- /dev/null +++ b/pkg/rpcserver/optional_params.go @@ -0,0 +1,10 @@ +package rpcserver + +import "github.com/filecoin-project/go-jsonrpc" + +func decodeOptionalParams(p jsonrpc.RawParams) ([]interface{}, error) { + if len(p) == 0 { + return nil, nil + } + return jsonrpc.DecodeParams[[]interface{}](p) +} diff --git a/pkg/rpcserver/parse_vote_account.go b/pkg/rpcserver/parse_vote_account.go new file mode 100644 index 000000000..6dd415536 --- /dev/null +++ b/pkg/rpcserver/parse_vote_account.go @@ -0,0 +1,127 @@ +package rpcserver + +import ( + "strconv" + + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" + "github.com/mr-tron/base58" +) + +type parsedVoteAccount struct { + Program string `json:"program"` + Parsed parsedVoteAccountBody `json:"parsed"` + Space uint64 `json:"space"` +} + +type parsedVoteAccountBody struct { + Type string `json:"type"` + Info parsedVoteAccountInfo `json:"info"` +} + +type parsedVoteAccountInfo struct { + NodePubkey string `json:"nodePubkey"` + AuthorizedWithdrawer string `json:"authorizedWithdrawer"` + Commission uint8 `json:"commission"` + Votes []parsedVote `json:"votes"` + RootSlot *uint64 `json:"rootSlot"` + AuthorizedVoters []parsedAuthorizedVoter `json:"authorizedVoters"` + PriorVoters []parsedPriorVoter `json:"priorVoters"` + EpochCredits []parsedEpochCredits `json:"epochCredits"` + LastTimestamp parsedBlockTimestamp `json:"lastTimestamp"` + InflationRewardsCommissionBPS uint16 `json:"inflationRewardsCommissionBps"` + InflationRewardsCollector string `json:"inflationRewardsCollector"` + BlockRevenueCollector string `json:"blockRevenueCollector"` + BlockRevenueCommissionBPS uint16 `json:"blockRevenueCommissionBps"` + PendingDelegatorRewards string `json:"pendingDelegatorRewards"` + BLSPubkeyCompressed *string `json:"blsPubkeyCompressed"` +} + +type parsedVote struct { + Latency uint8 `json:"latency"` + Slot uint64 `json:"slot"` + ConfirmationCount uint32 `json:"confirmationCount"` +} + +type parsedAuthorizedVoter struct { + Epoch uint64 `json:"epoch"` + AuthorizedVoter string `json:"authorizedVoter"` +} + +type parsedPriorVoter struct { + AuthorizedPubkey string `json:"authorizedPubkey"` + EpochOfLastAuthorizedSwitch uint64 `json:"epochOfLastAuthorizedSwitch"` + TargetEpoch uint64 `json:"targetEpoch"` +} + +type parsedEpochCredits struct { + Epoch uint64 `json:"epoch"` + Credits string `json:"credits"` + PreviousCredits string `json:"previousCredits"` +} + +type parsedBlockTimestamp struct { + Slot uint64 `json:"slot"` + Timestamp int64 `json:"timestamp"` +} + +func parsedVoteAccountData(data []byte, votePubkey solana.PublicKey) (parsedVoteAccount, error) { + versioned, err := sealevel.UnmarshalVersionedVoteState(data) + if err != nil { + return parsedVoteAccount{}, err + } + state := versioned.ConvertToCurrent() + info := parsedVoteAccountInfo{ + NodePubkey: state.NodePubkey.String(), + AuthorizedWithdrawer: state.AuthorizedWithdrawer.String(), + Commission: state.Commission, + Votes: make([]parsedVote, 0, state.Votes.Len()), + RootSlot: state.RootSlot, + AuthorizedVoters: make([]parsedAuthorizedVoter, 0, state.AuthorizedVoters.AuthorizedVoters.Len()), + PriorVoters: make([]parsedPriorVoter, 0), + EpochCredits: make([]parsedEpochCredits, 0, len(state.EpochCredits)), + LastTimestamp: parsedBlockTimestamp{ + Slot: state.LastTimestamp.Slot, Timestamp: state.LastTimestamp.Timestamp, + }, + InflationRewardsCommissionBPS: uint16(state.Commission) * 100, + InflationRewardsCollector: votePubkey.String(), + BlockRevenueCollector: state.NodePubkey.String(), + BlockRevenueCommissionBPS: 10000, + PendingDelegatorRewards: "0", + } + for i := 0; i < state.Votes.Len(); i++ { + vote := state.Votes.At(i) + info.Votes = append(info.Votes, parsedVote{ + Latency: vote.Latency, + Slot: vote.Lockout.Slot, + ConfirmationCount: vote.Lockout.ConfirmationCount, + }) + } + epochs, voters := state.AuthorizedVoters.AuthorizedVoters.KeyValues() + for i := range epochs { + info.AuthorizedVoters = append(info.AuthorizedVoters, parsedAuthorizedVoter{ + Epoch: epochs[i], AuthorizedVoter: voters[i].String(), + }) + } + for _, credits := range state.EpochCredits { + info.EpochCredits = append(info.EpochCredits, parsedEpochCredits{ + Epoch: credits.Epoch, Credits: strconv.FormatUint(credits.Credits, 10), PreviousCredits: strconv.FormatUint(credits.PrevCredits, 10), + }) + } + if versioned.Type == sealevel.VoteStateVersionV4 { + info.Commission, info.InflationRewardsCommissionBPS = voteCommission(versioned, state) + info.InflationRewardsCollector = versioned.V4.InflationRewardsCollector.String() + info.BlockRevenueCollector = versioned.V4.BlockRevenueCollector.String() + info.BlockRevenueCommissionBPS = versioned.V4.BlockRevenueCommissionBps + info.PendingDelegatorRewards = strconv.FormatUint(versioned.V4.PendingDelegatorRewards, 10) + if versioned.V4.BlsPubkeyCompressed != nil { + encoded := base58.Encode(versioned.V4.BlsPubkeyCompressed[:]) + info.BLSPubkeyCompressed = &encoded + } + } + return parsedVoteAccount{ + Program: "vote", + Parsed: parsedVoteAccountBody{Type: "vote", Info: info}, + Space: uint64(len(data)), + }, nil +} diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 8a3dd7afe..2a50624a0 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -11,8 +11,10 @@ import ( "net/http/httptest" "strings" "sync" + "sync/atomic" "time" + "github.com/Overclock-Validator/mithril/pkg/accounts" "github.com/Overclock-Validator/mithril/pkg/accountsdb" "github.com/Overclock-Validator/mithril/pkg/mlog" "github.com/Overclock-Validator/mithril/pkg/sealevel" @@ -31,6 +33,7 @@ type RpcServer struct { slotCtx *sealevel.SlotCtx slotCtxMu sync.RWMutex genesisHash string + rootedBank atomic.Pointer[rootedBankState] leaderTPUCacheMu sync.RWMutex leaderTPUByIdentity map[solana.PublicKey]tpuEndpoint @@ -45,19 +48,107 @@ type RpcServer struct { sendTransactionLeaderForwardCount uint64 } +// rootedBankState identifies the durable AccountsDB view served by state RPCs. +// Publishing it only after a successful fold keeps response context and account +// values on the same finalized bank. +type rootedBankState struct { + Slot uint64 + BlockHeight uint64 + TransactionCount uint64 +} + const maxQuietMethodProbeBody = 64 << 10 var supportedRPCMethods = map[string]struct{}{ "getAccountInfo": {}, + "getBalance": {}, "getBankHash": {}, + "getBlockProduction": {}, "getBlockHeight": {}, "getEpochInfo": {}, "getGenesisHash": {}, "getLatestBlockhash": {}, + "getLeaderSchedule": {}, + "getVoteAccounts": {}, "sendTransaction": {}, "simulateTransaction": {}, } +func (rpcServer *RpcServer) SetRootedBankState(slot, blockHeight, transactionCount uint64) { + rpcServer.rootedBank.Store(&rootedBankState{ + Slot: slot, + BlockHeight: blockHeight, + TransactionCount: transactionCount, + }) +} + +func (rpcServer *RpcServer) getRootedBankState() (rootedBankState, bool) { + rooted := rpcServer.rootedBank.Load() + if rooted == nil { + return rootedBankState{}, false + } + return *rooted, true +} + +func (rpcServer *RpcServer) readRootedAccount(ctx context.Context, pubkey solana.PublicKey) (rootedBankState, *accounts.Account, error) { + rooted, accountSet, err := rpcServer.readRootedAccounts(ctx, []solana.PublicKey{pubkey}) + if err != nil { + return rooted, nil, err + } + if len(accountSet) != 1 || accountSet[0] == nil { + return rooted, nil, accountsdb.ErrNoAccount + } + return rooted, accountSet[0], nil +} + +func (rpcServer *RpcServer) readRootedAccounts(ctx context.Context, pubkeys []solana.PublicKey) (rootedBankState, []*accounts.Account, error) { + for { + rooted, ok := rpcServer.getRootedBankState() + if !ok { + return rootedBankState{}, nil, fmt.Errorf("node has no rooted bank available") + } + if rpcServer.acctsDb == nil { + return rootedBankState{}, nil, fmt.Errorf("node has no accounts database available") + } + if rpcServer.rootedPublicationPending(rooted.Slot) { + if err := waitForRootedPublication(ctx); err != nil { + return rooted, nil, err + } + continue + } + accounts, stats, err := rpcServer.acctsDb.GetAccountsBatchSharedWithStats(ctx, rooted.Slot, pubkeys) + latest, stillPublished := rpcServer.getRootedBankState() + if stats.PendingFoldHits > 0 || !stillPublished || latest.Slot != rooted.Slot || + rpcServer.rootedPublicationPending(rooted.Slot) { + if err := waitForRootedPublication(ctx); err != nil { + return rooted, nil, err + } + continue + } + return rooted, accounts, err + } +} + +func (rpcServer *RpcServer) rootedPublicationPending(rootedSlot uint64) bool { + if !rpcServer.acctsDb.RootedDurable { + return false + } + durableThrough := rpcServer.acctsDb.DurableThrough() + // Zero means the snapshot baseline is active and no fold has committed yet. + return durableThrough != 0 && durableThrough != rootedSlot +} + +func waitForRootedPublication(ctx context.Context) error { + timer := time.NewTimer(time.Millisecond) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + func NewRpcServer(acctsDb *accountsdb.AccountsDb, port uint16, epochSchedule *sealevel.SysvarEpochSchedule, genesisHash solana.Hash) *RpcServer { var err error rpcServer := &RpcServer{genesisHash: genesisHash.String()} From 3cef30748f415849392835d0b17c7f8b9777e284 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Wed, 23 Sep 2026 23:30:30 +0530 Subject: [PATCH 2/6] test(rpc): cover vote accounts, balances, schedules, and production --- cmd/mithril/node/checkpoint_recovery_test.go | 28 +++ pkg/accountsdb/fold_test.go | 13 ++ pkg/rpcserver/get_account_info_parsed_test.go | 128 ++++++++++++ pkg/rpcserver/get_balance_test.go | 112 +++++++++++ pkg/rpcserver/get_block_production_test.go | 86 ++++++++ pkg/rpcserver/get_epoch_info_rooted_test.go | 33 ++++ pkg/rpcserver/get_leader_schedule_test.go | 83 ++++++++ pkg/rpcserver/get_vote_accounts_test.go | 185 ++++++++++++++++++ pkg/rpcserver/optional_params_test.go | 45 +++++ 9 files changed, 713 insertions(+) create mode 100644 pkg/rpcserver/get_account_info_parsed_test.go create mode 100644 pkg/rpcserver/get_balance_test.go create mode 100644 pkg/rpcserver/get_block_production_test.go create mode 100644 pkg/rpcserver/get_epoch_info_rooted_test.go create mode 100644 pkg/rpcserver/get_leader_schedule_test.go create mode 100644 pkg/rpcserver/get_vote_accounts_test.go create mode 100644 pkg/rpcserver/optional_params_test.go diff --git a/cmd/mithril/node/checkpoint_recovery_test.go b/cmd/mithril/node/checkpoint_recovery_test.go index d6ccc1f2b..ff5a1588c 100644 --- a/cmd/mithril/node/checkpoint_recovery_test.go +++ b/cmd/mithril/node/checkpoint_recovery_test.go @@ -134,6 +134,34 @@ func TestRecoveredManifestContextAdvancesStaleStateBeforeIntegrity(t *testing.T) assert.Equal(t, manifestCtx, s.LastRootedContext) } +type recordingRPCBankState struct { + slot uint64 + blockHeight uint64 + transactionCount uint64 +} + +func (state *recordingRPCBankState) SetRootedBankState(slot, blockHeight, transactionCount uint64) { + state.slot = slot + state.blockHeight = blockHeight + state.transactionCount = transactionCount +} + +func TestPublishRPCBankStateUsesRecoveredRoot(t *testing.T) { + transactionCount := uint64(789) + server := new(recordingRPCBankState) + publishRPCBankState(server, &state.MithrilState{ + ManifestParentSlot: 10, + ManifestBlockHeight: 20, + ManifestTransactionCount: 30, + LastRootedContext: &state.ResumeContext{ + Slot: 123, + BlockHeight: 456, + TransactionCount: &transactionCount, + }, + }) + assert.Equal(t, recordingRPCBankState{slot: 123, blockHeight: 456, transactionCount: 789}, *server) +} + func writeCheckpointManifest(t *testing.T, accountsDbPath, manifestsDir string, seq, root, fileID uint64, suffix string) *state.TransactionStatusCheckpointRef { t.Helper() payload := []byte(fmt.Sprintf("checkpoint-payload-%d", root)) diff --git a/pkg/accountsdb/fold_test.go b/pkg/accountsdb/fold_test.go index fbdff0ade..64f01d0f4 100644 --- a/pkg/accountsdb/fold_test.go +++ b/pkg/accountsdb/fold_test.go @@ -116,6 +116,19 @@ func TestCommitBatchReadsBackAndDedupes(t *testing.T) { assert.Equal(t, []byte("ctx-108"), manifest.ResumeCtx) } +func TestCommitBatchPublishesDurableThroughWithIndex(t *testing.T) { + db, _ := newFoldTestDb(t) + defer db.CloseDb() + + db.foldHooks.afterIndexCommit = func() { + require.Equal(t, uint64(42), db.DurableThrough()) + } + _, err := db.CommitBatch(foldDeltas( + accounts.SlotDelta{Slot: 42, Delta: []*accounts.Account{foldAcct(1, 1, nil)}}, + ), 42, nil, nil) + require.NoError(t, err) +} + // Manifest encode/decode round-trip, and CRC detection of a torn manifest. func TestSegmentManifestRoundTripAndTornDetection(t *testing.T) { dir := t.TempDir() diff --git a/pkg/rpcserver/get_account_info_parsed_test.go b/pkg/rpcserver/get_account_info_parsed_test.go new file mode 100644 index 000000000..9f9e7f08d --- /dev/null +++ b/pkg/rpcserver/get_account_info_parsed_test.go @@ -0,0 +1,128 @@ +package rpcserver + +import ( + "encoding/json" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestGetAccountInfoParsesVoteAccountAtRoot(t *testing.T) { + const rootedSlot = 500 + votePubkey := solana.PublicKey{1} + nodePubkey := solana.PublicKey{2} + withdrawer := solana.PublicKey{3} + collector := solana.PublicKey{4} + blockCollector := solana.PublicKey{5} + authorized := solana.PublicKey{6} + rootSlot := uint64(490) + bls := [48]byte{7} + state := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionV4} + state.V4.NodePubkey = nodePubkey + state.V4.AuthorizedWithdrawer = withdrawer + state.V4.InflationRewardsCollector = collector + state.V4.BlockRevenueCollector = blockCollector + state.V4.InflationRewardsCommissionBps = 755 + state.V4.BlockRevenueCommissionBps = 321 + state.V4.PendingDelegatorRewards = 42 + state.V4.BlsPubkeyCompressed = &bls + state.V4.RootSlot = &rootSlot + state.V4.AuthorizedVoters.AuthorizedVoters.Set(3, authorized) + state.V4.Votes.PushBack(sealevel.LandedVote{Latency: 2, Lockout: sealevel.VoteLockout{Slot: 499, ConfirmationCount: 4}}) + state.V4.EpochCredits = []sealevel.EpochCredits{{Epoch: 3, Credits: 9, PrevCredits: 5}} + state.V4.LastTimestamp = sealevel.BlockTimestamp{Slot: 499, Timestamp: 1_700_000_000} + + db := newRPCAccountsDB(t) + account := voteAccountForRPC(t, votePubkey, &state) + account.Lamports = 27_074_400 + _, err := db.CommitBatch([]accounts.SlotDelta{{Slot: rootedSlot, Delta: []*accounts.Account{account}}}, rootedSlot, nil, nil) + require.NoError(t, err) + + server := &RpcServer{acctsDb: db} + server.SetRootedBankState(rootedSlot, 495, 100) + got, err := server.GetAccountInfo(t.Context(), mustRawParams(t, []interface{}{ + votePubkey.String(), + map[string]interface{}{"commitment": "finalized", "encoding": "jsonParsed", "minContextSlot": float64(rootedSlot)}, + })) + require.NoError(t, err) + require.Equal(t, uint64(rootedSlot), got.Context.Slot) + require.NotNil(t, got.Value) + require.Equal(t, uint64(27_074_400), got.Value.Lamports) + + raw, err := json.Marshal(got.Value.Data) + require.NoError(t, err) + require.Contains(t, string(raw), `"lastTimestamp":{"slot":499,"timestamp":1700000000}`) + var decoded struct { + Program string `json:"program"` + Parsed struct { + Type string `json:"type"` + Info parsedVoteAccountInfo `json:"info"` + } `json:"parsed"` + Space uint64 `json:"space"` + } + require.NoError(t, json.Unmarshal(raw, &decoded)) + require.Equal(t, "vote", decoded.Program) + require.Equal(t, "vote", decoded.Parsed.Type) + require.Equal(t, nodePubkey.String(), decoded.Parsed.Info.NodePubkey) + require.Equal(t, withdrawer.String(), decoded.Parsed.Info.AuthorizedWithdrawer) + require.Equal(t, uint8(8), decoded.Parsed.Info.Commission) + require.Equal(t, uint16(755), decoded.Parsed.Info.InflationRewardsCommissionBPS) + require.Equal(t, "42", decoded.Parsed.Info.PendingDelegatorRewards) + require.Equal(t, parsedBlockTimestamp{Slot: 499, Timestamp: 1_700_000_000}, decoded.Parsed.Info.LastTimestamp) + require.Equal(t, []parsedAuthorizedVoter{{Epoch: 3, AuthorizedVoter: authorized.String()}}, decoded.Parsed.Info.AuthorizedVoters) + require.Equal(t, []parsedVote{{Latency: 2, Slot: 499, ConfirmationCount: 4}}, decoded.Parsed.Info.Votes) + require.Equal(t, []parsedEpochCredits{{Epoch: 3, Credits: "9", PreviousCredits: "5"}}, decoded.Parsed.Info.EpochCredits) +} + +func TestGetAccountInfoJsonParsedFallsBackToBase64(t *testing.T) { + db := newRPCAccountsDB(t) + address := solana.PublicKey{9} + _, err := db.CommitBatch([]accounts.SlotDelta{{Slot: 10, Delta: []*accounts.Account{{Key: address, Lamports: 1, Data: []byte{1, 2, 3}}}}}, 10, nil, nil) + require.NoError(t, err) + server := &RpcServer{acctsDb: db} + server.SetRootedBankState(10, 9, 0) + + got, err := server.GetAccountInfo(t.Context(), mustRawParams(t, []interface{}{ + address.String(), map[string]interface{}{"encoding": "jsonParsed"}, + })) + require.NoError(t, err) + require.Equal(t, []string{"AQID", "base64"}, got.Value.Data) +} + +func TestGetAccountInfoParsesLegacyVoteAccountWithV4Defaults(t *testing.T) { + votePubkey := solana.PublicKey{1} + nodePubkey := solana.PublicKey{2} + state := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionCurrent} + state.Current.NodePubkey = nodePubkey + state.Current.Commission = 7 + + db := newRPCAccountsDB(t) + _, err := db.CommitBatch([]accounts.SlotDelta{{Slot: 10, Delta: []*accounts.Account{ + voteAccountForRPC(t, votePubkey, &state), + }}}, 10, nil, nil) + require.NoError(t, err) + server := &RpcServer{acctsDb: db} + server.SetRootedBankState(10, 9, 0) + + got, err := server.GetAccountInfo(t.Context(), mustRawParams(t, []interface{}{ + votePubkey.String(), map[string]interface{}{"encoding": "jsonParsed"}, + })) + require.NoError(t, err) + raw, err := json.Marshal(got.Value.Data) + require.NoError(t, err) + var decoded struct { + Parsed struct { + Info parsedVoteAccountInfo `json:"info"` + } `json:"parsed"` + } + require.NoError(t, json.Unmarshal(raw, &decoded)) + require.Equal(t, uint8(7), decoded.Parsed.Info.Commission) + require.Equal(t, uint16(700), decoded.Parsed.Info.InflationRewardsCommissionBPS) + require.Equal(t, votePubkey.String(), decoded.Parsed.Info.InflationRewardsCollector) + require.Equal(t, nodePubkey.String(), decoded.Parsed.Info.BlockRevenueCollector) + require.Equal(t, uint16(10000), decoded.Parsed.Info.BlockRevenueCommissionBPS) + require.Equal(t, "0", decoded.Parsed.Info.PendingDelegatorRewards) +} diff --git a/pkg/rpcserver/get_balance_test.go b/pkg/rpcserver/get_balance_test.go new file mode 100644 index 000000000..04ae95c97 --- /dev/null +++ b/pkg/rpcserver/get_balance_test.go @@ -0,0 +1,112 @@ +package rpcserver + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + "github.com/Overclock-Validator/mithril/pkg/accountsdb" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestGetBalanceUsesPublishedRootedBank(t *testing.T) { + db := newRPCAccountsDB(t) + address := solana.PublicKey{7} + _, err := db.CommitBatch([]accounts.SlotDelta{{ + Slot: 42, + Delta: []*accounts.Account{{ + Key: address, + Lamports: 123_456_789, + }}, + }}, 42, nil, nil) + require.NoError(t, err) + + server := &RpcServer{acctsDb: db} + server.SetRootedBankState(42, 40, 10) + got, err := server.GetBalance(t.Context(), mustRawParams(t, []interface{}{ + address.String(), + map[string]interface{}{"commitment": "confirmed", "minContextSlot": float64(42)}, + })) + require.NoError(t, err) + require.Equal(t, uint64(42), got.Context.Slot) + require.Equal(t, uint64(123_456_789), got.Value) + + _, err = server.GetBalance(t.Context(), mustRawParams(t, []interface{}{ + address.String(), + map[string]interface{}{"minContextSlot": float64(43)}, + })) + var minSlotErr *MinContextSlotNotReachedError + require.True(t, errors.As(err, &minSlotErr)) + require.Equal(t, uint64(42), minSlotErr.ContextSlot) +} + +func TestGetBalanceMissingAccountIsZero(t *testing.T) { + server := &RpcServer{acctsDb: newRPCAccountsDB(t)} + server.SetRootedBankState(9, 8, 0) + + got, err := server.GetBalance(t.Context(), mustRawParams(t, []interface{}{solana.PublicKey{8}.String()})) + require.NoError(t, err) + require.Equal(t, uint64(9), got.Context.Slot) + require.Zero(t, got.Value) +} + +func TestGetBalanceDoesNotMixNewAccountsWithOldRoot(t *testing.T) { + db := newRPCAccountsDB(t) + address := solana.PublicKey{9} + _, err := db.CommitBatch([]accounts.SlotDelta{{ + Slot: 43, + Delta: []*accounts.Account{{ + Key: address, + Lamports: 43, + }}, + }}, 43, nil, nil) + require.NoError(t, err) + + server := &RpcServer{acctsDb: db} + server.SetRootedBankState(42, 40, 10) + deadline, cancel := context.WithTimeout(t.Context(), 5*time.Millisecond) + defer cancel() + _, err = server.GetBalance(deadline, mustRawParams(t, []interface{}{address.String()})) + require.ErrorIs(t, err, context.DeadlineExceeded) + + server.SetRootedBankState(43, 41, 11) + got, err := server.GetBalance(t.Context(), mustRawParams(t, []interface{}{address.String()})) + require.NoError(t, err) + require.Equal(t, uint64(43), got.Context.Slot) + require.Equal(t, uint64(43), got.Value) +} + +func TestGetBalanceRejectsInvalidParams(t *testing.T) { + server := &RpcServer{} + for _, params := range [][]interface{}{ + {}, + {true}, + {"not-a-pubkey"}, + {solana.PublicKey{1}.String(), true}, + {solana.PublicKey{1}.String(), map[string]interface{}{"commitment": "unknown"}}, + {solana.PublicKey{1}.String(), map[string]interface{}{"minContextSlot": 1.5}}, + } { + _, err := server.GetBalance(t.Context(), mustRawParams(t, params)) + var invalid *InvalidParamsError + require.True(t, errors.As(err, &invalid), "params: %#v; error: %v", params, err) + } +} + +func newRPCAccountsDB(t *testing.T) *accountsdb.AccountsDb { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(dir, "accounts"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "largest_file_id"), make([]byte, 8), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "bootstrap_high_file_id"), make([]byte, 8), 0o644)) + db, err := accountsdb.OpenDb(dir) + require.NoError(t, err) + db.RootedDurable = true + db.InitCaches() + t.Cleanup(db.CloseDb) + return db +} diff --git a/pkg/rpcserver/get_block_production_test.go b/pkg/rpcserver/get_block_production_test.go new file mode 100644 index 000000000..19e7e396c --- /dev/null +++ b/pkg/rpcserver/get_block_production_test.go @@ -0,0 +1,86 @@ +package rpcserver + +import ( + "errors" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/leaderschedule" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestGetBlockProductionUsesRootedSlotHistory(t *testing.T) { + const rootedSlot = uint64(13) + firstLeader := solana.PublicKey{1} + secondLeader := solana.PublicKey{2} + global.SetLeaderScheduleForEpoch(0, leaderschedule.NewLeaderScheduleFromKeyedSlots( + map[solana.PublicKey][]uint64{ + firstLeader: {10, 11}, + secondLeader: {12, 13}, + }, + 0, + )) + t.Cleanup(func() { global.SetLeaderScheduleForEpoch(0, nil) }) + + history := sealevel.SysvarSlotHistory{ + Bits: sealevel.SlotHistoryBitvec{ + Bits: sealevel.SlotHistoryInner{BlocksLen: 1, Blocks: []uint64{(1 << 10) | (1 << 12) | (1 << 13)}}, + Len: 64, + }, + NextSlot: rootedSlot + 1, + } + db := newRPCAccountsDB(t) + _, err := db.CommitBatch([]accounts.SlotDelta{{ + Slot: rootedSlot, + Delta: []*accounts.Account{{ + Key: sealevel.SysvarSlotHistoryAddr, + Lamports: 1, + Data: history.MustMarshal(), + }}, + }}, rootedSlot, nil, nil) + require.NoError(t, err) + + server := &RpcServer{ + acctsDb: db, + epochSchedule: &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 32}, + } + server.SetRootedBankState(rootedSlot, 12, 0) + got, err := server.GetBlockProduction(t.Context(), mustRawParams(t, []interface{}{ + map[string]interface{}{ + "commitment": "finalized", + "range": map[string]interface{}{ + "firstSlot": float64(10), + "lastSlot": float64(rootedSlot), + }, + }, + })) + require.NoError(t, err) + require.Equal(t, rootedSlot, got.Context.Slot) + require.Equal(t, BlockProductionRange{FirstSlot: 10, LastSlot: 13}, got.Value.Range) + require.Equal(t, [2]uint64{2, 1}, got.Value.ByIdentity[firstLeader.String()]) + require.Equal(t, [2]uint64{2, 2}, got.Value.ByIdentity[secondLeader.String()]) +} + +func TestGetBlockProductionRejectsInvalidRange(t *testing.T) { + server := &RpcServer{ + acctsDb: newRPCAccountsDB(t), + epochSchedule: &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 32}, + } + server.SetRootedBankState(13, 12, 0) + for _, params := range [][]interface{}{ + {true}, + {map[string]interface{}{"commitment": "unknown"}}, + {map[string]interface{}{"identity": "invalid"}}, + {map[string]interface{}{"range": true}}, + {map[string]interface{}{"range": map[string]interface{}{}}}, + {map[string]interface{}{"range": map[string]interface{}{"firstSlot": 1.5}}}, + {map[string]interface{}{}, map[string]interface{}{}}, + } { + _, err := server.GetBlockProduction(t.Context(), mustRawParams(t, params)) + var invalid *InvalidParamsError + require.True(t, errors.As(err, &invalid), "params: %#v; error: %v", params, err) + } +} diff --git a/pkg/rpcserver/get_epoch_info_rooted_test.go b/pkg/rpcserver/get_epoch_info_rooted_test.go new file mode 100644 index 000000000..0d1336411 --- /dev/null +++ b/pkg/rpcserver/get_epoch_info_rooted_test.go @@ -0,0 +1,33 @@ +package rpcserver + +import ( + "testing" + + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/filecoin-project/go-jsonrpc" + "github.com/stretchr/testify/require" +) + +func TestGetEpochInfoUsesRootedBankForFinalized(t *testing.T) { + oldSlot, oldEpoch := global.Slot(), global.Epoch() + oldHeight, oldTransactions := global.BlockHeight(), global.TransactionCount() + t.Cleanup(func() { + global.SetSlot(oldSlot) + global.SetEpoch(oldEpoch) + global.SetBlockHeight(oldHeight) + global.SetTransactionCount(oldTransactions) + }) + global.SetSlot(432250) + global.SetEpoch(1) + global.SetBlockHeight(432000) + global.SetTransactionCount(2000) + + server := &RpcServer{epochSchedule: &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 432000, LeaderScheduleSlotOffset: 432000}} + server.SetRootedBankState(432123, 431900, 1900) + got, err := server.GetEpochInfo(t.Context(), jsonrpc.RawParams(`[{"commitment":"finalized"}]`)) + require.NoError(t, err) + require.Equal(t, uint64(432123), got.AbsoluteSlot) + require.Equal(t, uint64(431900), got.BlockHeight) + require.Equal(t, uint64(1900), got.TransactionCount) +} diff --git a/pkg/rpcserver/get_leader_schedule_test.go b/pkg/rpcserver/get_leader_schedule_test.go new file mode 100644 index 000000000..836924bb0 --- /dev/null +++ b/pkg/rpcserver/get_leader_schedule_test.go @@ -0,0 +1,83 @@ +package rpcserver + +import ( + "errors" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/leaderschedule" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestGetLeaderScheduleReturnsRelativeEpochSlots(t *testing.T) { + epochSchedule := &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 4} + firstSlot := epochSchedule.FirstSlotInEpoch(3) + firstLeader := solana.PublicKey{1} + secondLeader := solana.PublicKey{2} + global.SetLeaderScheduleForEpoch(3, leaderschedule.NewLeaderScheduleFromKeyedSlots( + map[solana.PublicKey][]uint64{ + firstLeader: {0, 2}, + secondLeader: {1, 3}, + }, + firstSlot, + )) + t.Cleanup(func() { global.SetLeaderScheduleForEpoch(3, nil) }) + + server := &RpcServer{epochSchedule: epochSchedule} + server.SetRootedBankState(firstSlot+2, firstSlot+2, 0) + got, err := server.GetLeaderSchedule(t.Context(), mustRawParams(t, []interface{}{ + float64(firstSlot + 1), + map[string]interface{}{"commitment": "confirmed"}, + })) + require.NoError(t, err) + require.Equal(t, []uint64{0, 2}, got[firstLeader.String()]) + require.Equal(t, []uint64{1, 3}, got[secondLeader.String()]) + + filtered, err := server.GetLeaderSchedule(t.Context(), mustRawParams(t, []interface{}{ + float64(firstSlot), + map[string]interface{}{"identity": secondLeader.String()}, + })) + require.NoError(t, err) + require.Equal(t, map[string][]uint64{secondLeader.String(): {1, 3}}, filtered) + + filtered, err = server.GetLeaderSchedule(t.Context(), mustRawParams(t, []interface{}{ + map[string]interface{}{"commitment": "finalized", "identity": secondLeader.String()}, + })) + require.NoError(t, err) + require.Equal(t, map[string][]uint64{secondLeader.String(): {1, 3}}, filtered) + + missing, err := server.GetLeaderSchedule(t.Context(), mustRawParams(t, []interface{}{ + float64(firstSlot), map[string]interface{}{"identity": solana.PublicKey{9}.String()}, + })) + require.NoError(t, err) + require.Empty(t, missing) +} + +func TestGetLeaderScheduleReturnsNullWhenEpochScheduleIsUnavailable(t *testing.T) { + server := &RpcServer{epochSchedule: &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 4}} + server.SetRootedBankState(40_002, 1, 0) + + got, err := server.GetLeaderSchedule(t.Context(), mustRawParams(t, []interface{}{float64(40_002)})) + require.NoError(t, err) + require.Nil(t, got) +} + +func TestGetLeaderScheduleRejectsInvalidParams(t *testing.T) { + server := &RpcServer{epochSchedule: &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 4}} + server.SetRootedBankState(1, 1, 0) + for _, params := range [][]interface{}{ + {-1.0}, + {1.5}, + {float64(1), true}, + {float64(1), map[string]interface{}{"commitment": "unknown"}}, + {float64(1), map[string]interface{}{"identity": "invalid"}}, + {float64(1), map[string]interface{}{"keyByVoteAccount": true}}, + {float64(1), nil, nil}, + } { + _, err := server.GetLeaderSchedule(t.Context(), mustRawParams(t, params)) + var invalid *InvalidParamsError + require.True(t, errors.As(err, &invalid), "params: %#v; error: %v", params, err) + } +} diff --git a/pkg/rpcserver/get_vote_accounts_test.go b/pkg/rpcserver/get_vote_accounts_test.go new file mode 100644 index 000000000..d6f9d35c3 --- /dev/null +++ b/pkg/rpcserver/get_vote_accounts_test.go @@ -0,0 +1,185 @@ +package rpcserver + +import ( + "bytes" + "errors" + "math" + "path/filepath" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + "github.com/Overclock-Validator/mithril/pkg/addresses" + "github.com/Overclock-Validator/mithril/pkg/epochstakes" + "github.com/Overclock-Validator/mithril/pkg/features" + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + bin "github.com/gagliardetto/binary" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestGetVoteAccountsUsesRootedVoteState(t *testing.T) { + const ( + rootedSlot = 500 + epoch = 0 + ) + currentVote := solana.PublicKey{1} + delinquentVote := solana.PublicKey{2} + newVote := solana.PublicKey{3} + currentNode := solana.PublicKey{11} + delinquentNode := solana.PublicKey{12} + newNode := solana.PublicKey{13} + currentRoot := uint64(498) + delinquentRoot := uint64(299) + + currentState := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionV4} + currentState.V4.NodePubkey = currentNode + currentState.V4.InflationRewardsCommissionBps = 755 + currentState.V4.RootSlot = ¤tRoot + currentState.V4.Votes.PushBack(sealevel.LandedVote{Lockout: sealevel.VoteLockout{Slot: 499}}) + currentState.V4.EpochCredits = []sealevel.EpochCredits{ + {Epoch: 0, Credits: 1}, + {Epoch: 1, Credits: 2, PrevCredits: 1}, + {Epoch: 2, Credits: 3, PrevCredits: 2}, + {Epoch: 3, Credits: 4, PrevCredits: 3}, + {Epoch: 4, Credits: 5, PrevCredits: 4}, + {Epoch: 5, Credits: 9, PrevCredits: 5}, + } + + delinquentState := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionCurrent} + delinquentState.Current.NodePubkey = delinquentNode + delinquentState.Current.Commission = 5 + delinquentState.Current.RootSlot = &delinquentRoot + delinquentState.Current.Votes.PushBack(sealevel.LandedVote{Lockout: sealevel.VoteLockout{Slot: 300}}) + + newState := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionCurrent} + newState.Current.NodePubkey = newNode + newState.Current.Votes.PushBack(sealevel.LandedVote{Lockout: sealevel.VoteLockout{Slot: 498}}) + + db := newRPCAccountsDB(t) + global.ClearPendingStakePubkeys() + t.Cleanup(func() { + global.EnqueuePendingStakePubkey(rootedSlot, solana.PublicKey{21}) + _, err := global.FlushPendingStakePubkeys(filepath.Join(db.AcctsDir, "..")) + require.NoError(t, err) + global.ClearPendingStakePubkeys() + }) + stakeHistory := sealevel.SysvarStakeHistory{} + var historyData bytes.Buffer + require.NoError(t, stakeHistory.MarshalWithEncoder(bin.NewBinEncoder(&historyData))) + stakeAccount := func(key, vote solana.PublicKey, lamports uint64) *accounts.Account { + state := &sealevel.StakeStateV2{Status: sealevel.StakeStateV2StatusStake} + state.Stake.Stake.Delegation = sealevel.Delegation{ + VoterPubkey: vote, StakeLamports: lamports, + ActivationEpoch: math.MaxUint64, DeactivationEpoch: math.MaxUint64, + } + data, err := sealevel.MarshalStakeStake(state) + require.NoError(t, err) + global.EnqueuePendingStakePubkey(rootedSlot, key) + return &accounts.Account{Key: key, Owner: addresses.StakeProgramAddr, Lamports: lamports, Data: data} + } + _, err := db.CommitBatch([]accounts.SlotDelta{{ + Slot: rootedSlot, + Delta: []*accounts.Account{ + voteAccountForRPC(t, currentVote, ¤tState), + voteAccountForRPC(t, delinquentVote, &delinquentState), + voteAccountForRPC(t, newVote, &newState), + stakeAccount(solana.PublicKey{21}, currentVote, 125), + stakeAccount(solana.PublicKey{22}, delinquentVote, 200), + stakeAccount(solana.PublicKey{23}, solana.PublicKey{4}, 50), + {Key: sealevel.SysvarStakeHistoryAddr, Owner: addresses.SysvarOwnerAddr, Lamports: 1, Data: historyData.Bytes()}, + }, + }}, rootedSlot, nil, nil) + require.NoError(t, err) + _, err = global.FlushPendingStakePubkeys(filepath.Join(db.AcctsDir, "..")) + require.NoError(t, err) + + previous, hadPrevious := global.EpochStakesSnapshot(epoch) + global.PutEpochStakes(epoch, + map[solana.PublicKey]uint64{currentVote: 100, delinquentVote: 200}, + map[solana.PublicKey]*epochstakes.VoteAccount{ + currentVote: {NodePubkey: currentNode}, + delinquentVote: {NodePubkey: delinquentNode}, + }, + 300, + ) + t.Cleanup(func() { + if hadPrevious { + global.PutEpochStakes(epoch, previous.Stakes, previous.VoteAccounts, previous.TotalStake) + } else { + global.ClearEpochStakes(epoch) + } + }) + + server := &RpcServer{ + acctsDb: db, + epochSchedule: &sealevel.SysvarEpochSchedule{ + SlotsPerEpoch: 1_000, + }, + } + server.SetSlotCtx(&sealevel.SlotCtx{Features: features.NewFeaturesDefault()}) + server.SetRootedBankState(rootedSlot, 490, 1_000) + got, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{ + map[string]interface{}{"commitment": "confirmed"}, + })) + require.NoError(t, err) + require.Len(t, got.Current, 2) + require.Len(t, got.Delinquent, 1) + require.Equal(t, currentVote.String(), got.Current[0].VotePubkey) + require.Equal(t, currentNode.String(), got.Current[0].NodePubkey) + require.Equal(t, uint64(125), got.Current[0].ActivatedStake, "live stake differs from frozen epoch stake") + require.Equal(t, uint8(8), got.Current[0].Commission) + require.Equal(t, uint16(755), got.Current[0].InflationRewardsCommissionBPS) + require.Equal(t, uint64(499), got.Current[0].LastVote) + require.Equal(t, [][3]uint64{{1, 2, 1}, {2, 3, 2}, {3, 4, 3}, {4, 5, 4}, {5, 9, 5}}, got.Current[0].EpochCredits) + require.True(t, got.Current[0].EpochVoteAccount) + require.Equal(t, newVote.String(), got.Current[1].VotePubkey) + require.Equal(t, uint64(0), got.Current[1].ActivatedStake) + require.False(t, got.Current[1].EpochVoteAccount) + require.Equal(t, delinquentVote.String(), got.Delinquent[0].VotePubkey) + require.Equal(t, uint8(5), got.Delinquent[0].Commission) + + filtered, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{ + map[string]interface{}{"votePubkey": currentVote.String()}, + })) + require.NoError(t, err) + require.Len(t, filtered.Current, 1) + require.Empty(t, filtered.Delinquent) + + db.VoteAcctCache.Clear() + cold, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{})) + require.NoError(t, err) + require.Len(t, cold.Current, 1) + require.Equal(t, currentVote.String(), cold.Current[0].VotePubkey) + require.Len(t, cold.Delinquent, 1) + require.Equal(t, delinquentVote.String(), cold.Delinquent[0].VotePubkey) +} + +func TestGetVoteAccountsRejectsInvalidConfig(t *testing.T) { + server := &RpcServer{} + for _, params := range [][]interface{}{ + {true}, + {map[string]interface{}{"commitment": "unknown"}}, + {map[string]interface{}{"votePubkey": "invalid"}}, + {map[string]interface{}{"keepUnstakedDelinquents": "yes"}}, + {map[string]interface{}{"keepUnstakedDelinquents": true}}, + {map[string]interface{}{"delinquentSlotDistance": 1.5}}, + {map[string]interface{}{}, map[string]interface{}{}}, + } { + _, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, params)) + var invalid *InvalidParamsError + require.True(t, errors.As(err, &invalid), "params: %#v; error: %v", params, err) + } +} + +func voteAccountForRPC(t *testing.T, key solana.PublicKey, state *sealevel.VoteStateVersions) *accounts.Account { + t.Helper() + data, err := sealevel.MarshalVersionedVoteState(state) + require.NoError(t, err) + return &accounts.Account{ + Key: key, + Lamports: 1, + Owner: addresses.VoteProgramAddr, + Data: data, + } +} diff --git a/pkg/rpcserver/optional_params_test.go b/pkg/rpcserver/optional_params_test.go new file mode 100644 index 000000000..b0cfa31a1 --- /dev/null +++ b/pkg/rpcserver/optional_params_test.go @@ -0,0 +1,45 @@ +package rpcserver + +import ( + "context" + "testing" + + "github.com/filecoin-project/go-jsonrpc" + "github.com/stretchr/testify/require" +) + +func TestOptionalRPCParamsCanBeOmitted(t *testing.T) { + server := &RpcServer{} + methods := map[string]func(context.Context, jsonrpc.RawParams) error{ + "getEpochInfo": func(ctx context.Context, p jsonrpc.RawParams) error { + _, err := server.GetEpochInfo(ctx, p) + return err + }, + "getVoteAccounts": func(ctx context.Context, p jsonrpc.RawParams) error { + _, err := server.GetVoteAccounts(ctx, p) + return err + }, + "getLeaderSchedule": func(ctx context.Context, p jsonrpc.RawParams) error { + _, err := server.GetLeaderSchedule(ctx, p) + return err + }, + "getBlockProduction": func(ctx context.Context, p jsonrpc.RawParams) error { + _, err := server.GetBlockProduction(ctx, p) + return err + }, + } + for name, call := range methods { + t.Run(name, func(t *testing.T) { + // Missing node state should fail identically for each empty parameter form. + want := call(t.Context(), jsonrpc.RawParams(`[]`)) + require.Error(t, want) + for _, params := range []jsonrpc.RawParams{nil, jsonrpc.RawParams(`null`)} { + require.EqualError(t, call(t.Context(), params), want.Error()) + } + for _, params := range []jsonrpc.RawParams{jsonrpc.RawParams(`{}`), jsonrpc.RawParams(`true`)} { + var invalid *InvalidParamsError + require.ErrorAs(t, call(t.Context(), params), &invalid) + } + }) + } +} From bd11a01a3b9c9fa3252dfd25d2c23c702e8a3581 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Thu, 24 Sep 2026 13:10:48 +0530 Subject: [PATCH 3/6] rpc: persist vote-account candidates and use rooted feature state --- pkg/accountsdb/accountsdb.go | 11 +-- pkg/accountsdb/compact.go | 2 +- pkg/accountsdb/fold.go | 11 ++- pkg/accountsdb/index.go | 11 ++- pkg/accountsdb/segment.go | 16 ++- pkg/accountsdb/vote_index.go | 153 +++++++++++++++++++++++++++++ pkg/rpcserver/get_vote_accounts.go | 38 ++++--- pkg/snapshot/build_db.go | 52 +++++----- pkg/snapshot/build_db_with_incr.go | 14 ++- 9 files changed, 246 insertions(+), 62 deletions(-) create mode 100644 pkg/accountsdb/vote_index.go diff --git a/pkg/accountsdb/accountsdb.go b/pkg/accountsdb/accountsdb.go index 4c1fd2239..f738d5c23 100644 --- a/pkg/accountsdb/accountsdb.go +++ b/pkg/accountsdb/accountsdb.go @@ -32,6 +32,7 @@ type AccountsDb struct { AcctsDir string LargestFileId atomic.Uint64 VoteAcctCache otter.Cache[solana.PublicKey, *accounts.Account] + voteIndexMu sync.Mutex CommonAcctsCache otter.Cache[solana.PublicKey, *accounts.Account] ProgramCache otter.Cache[solana.PublicKey, *ProgramCacheEntry] // Otter permits concurrent ordinary operations but not Clear. Rewind takes @@ -102,16 +103,6 @@ func (accountsDb *AccountsDb) DurableThrough() uint64 { return accountsDb.durableThrough.Load() } -// VoteAccountPubkeys returns the vote accounts in the current cache view. -func (accountsDb *AccountsDb) VoteAccountPubkeys() []solana.PublicKey { - pubkeys := make([]solana.PublicKey, 0, accountsDb.VoteAcctCache.Size()) - accountsDb.VoteAcctCache.Range(func(pubkey solana.PublicKey, _ *accounts.Account) bool { - pubkeys = append(pubkeys, pubkey) - return true - }) - return pubkeys -} - // silentLogger implements pebble.Logger but discards all messages. // This suppresses verbose WAL recovery messages on startup. type silentLogger struct{} diff --git a/pkg/accountsdb/compact.go b/pkg/accountsdb/compact.go index 6b167f7ed..9e2f61481 100644 --- a/pkg/accountsdb/compact.go +++ b/pkg/accountsdb/compact.go @@ -268,7 +268,7 @@ func (db *AccountsDb) compactFile(c compactCandidate, minDeadFraction float64) ( return false, 0, err } - pubkeys, idxEntries, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), c.slot, c.fileId) + pubkeys, idxEntries, _, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), c.slot, c.fileId) if err != nil { return false, 0, err } diff --git a/pkg/accountsdb/fold.go b/pkg/accountsdb/fold.go index 7e01dfe8d..49391695b 100644 --- a/pkg/accountsdb/fold.go +++ b/pkg/accountsdb/fold.go @@ -13,6 +13,7 @@ import ( "sort" "github.com/Overclock-Validator/mithril/pkg/accounts" + "github.com/Overclock-Validator/mithril/pkg/addresses" "github.com/cockroachdb/pebble" "golang.org/x/sync/errgroup" ) @@ -181,7 +182,10 @@ func (db *AccountsDb) CommitBatch( segErr := func() error { for _, k := range keys { v := union[k] - records = append(records, ManifestRecord{Pubkey: k, Offset: dataLen, OwnerSlot: v.ownerSlot}) + records = append(records, ManifestRecord{ + Pubkey: k, Offset: dataLen, OwnerSlot: v.ownerSlot, + Vote: v.acct.Lamports > 0 && v.acct.Owner == addresses.VoteProgramAddr, + }) ava := AppendVecAccount{ DataLen: uint64(len(v.acct.Data)), Pubkey: v.acct.Key, @@ -339,6 +343,11 @@ func (db *AccountsDb) applyManifestToIndex(m *SegmentManifest) error { if err := batch.Set(r.Pubkey[:], idxBuf[:], nil); err != nil { return err } + if r.Vote { + if err := batch.Set(voteIndexKey(r.Pubkey), nil, nil); err != nil { + return err + } + } } if err := batch.Set(metaKeyLastBatch, encodeFoldMeta(foldMeta{ BatchSeq: m.BatchSeq, diff --git a/pkg/accountsdb/index.go b/pkg/accountsdb/index.go index 7601d73f4..0a064517a 100644 --- a/pkg/accountsdb/index.go +++ b/pkg/accountsdb/index.go @@ -100,10 +100,12 @@ func WriteStakePubkeyIndex(path string, entries []StakeIndexEntry) error { // - pubkeys: all account pubkeys // - acctIdxEntries: index entries for each account // - stakeEntries: stake account pubkeys with their appendvec location hints -func BuildIndexEntriesFromAppendVecs(data []byte, fileSize uint64, slot uint64, fileId uint64) ([]solana.PublicKey, []AccountIndexEntry, []StakeIndexEntry, error) { +// - votePubkeys: vote-program account candidates, including accounts without active stake +func BuildIndexEntriesFromAppendVecs(data []byte, fileSize uint64, slot uint64, fileId uint64) ([]solana.PublicKey, []AccountIndexEntry, []StakeIndexEntry, []solana.PublicKey, error) { pubkeys := make([]solana.PublicKey, 0, 20000) acctIdxEntries := make([]AccountIndexEntry, 0, 20000) stakeEntries := make([]StakeIndexEntry, 0, 1000) + var votePubkeys []solana.PublicKey parser := &appendVecParser{Buf: data, FileSize: fileSize, FileId: fileId, Slot: slot} var owner solana.PublicKey @@ -117,7 +119,7 @@ func BuildIndexEntriesFromAppendVecs(data []byte, fileSize uint64, slot uint64, if errors.Is(err, io.EOF) { break } - return nil, nil, nil, fmt.Errorf("parse appendvec slot=%d file_id=%d: %w", slot, fileId, err) + return nil, nil, nil, nil, fmt.Errorf("parse appendvec slot=%d file_id=%d: %w", slot, fileId, err) } // Collect stake account entries with appendvec location hints if bytes.Equal(owner[:], addresses.StakeProgramAddr[:]) { @@ -128,7 +130,10 @@ func BuildIndexEntriesFromAppendVecs(data []byte, fileSize uint64, slot uint64, Offset: acctIdxEntries[idx].Offset, }) } + if owner == addresses.VoteProgramAddr { + votePubkeys = append(votePubkeys, pubkeys[len(pubkeys)-1]) + } } - return pubkeys, acctIdxEntries, stakeEntries, nil + return pubkeys, acctIdxEntries, stakeEntries, votePubkeys, nil } diff --git a/pkg/accountsdb/segment.go b/pkg/accountsdb/segment.go index 9359f7841..5cc189f8a 100644 --- a/pkg/accountsdb/segment.go +++ b/pkg/accountsdb/segment.go @@ -62,6 +62,7 @@ type ManifestRecord struct { Offset uint64 // record offset within this segment's data file OwnerSlot uint64 // slot that produced this version (observability) PrevValid bool // false => the key was absent from the index before this batch + Vote bool // this version is a nonzero-lamport vote-program account Prev AccountIndexEntry // index entry before this batch overwrote it } @@ -138,11 +139,14 @@ func (m *SegmentManifest) encode() []byte { body.Write(r.Pubkey[:]) put64(r.Offset) put64(r.OwnerSlot) + var flags byte if r.PrevValid { - body.WriteByte(1) - } else { - body.WriteByte(0) + flags |= 1 } + if r.Vote { + flags |= 2 + } + body.WriteByte(flags) r.Prev.Marshal(&prevBuf) body.Write(prevBuf[:]) } @@ -324,7 +328,11 @@ func ReadSegmentManifest(path string) (*SegmentManifest, error) { if err != nil { return nil, err } - r.PrevValid = pv == 1 + if pv&^3 != 0 { + return nil, ErrTornManifest + } + r.PrevValid = pv&1 != 0 + r.Vote = pv&2 != 0 prevRaw, err := d.bytes(24) if err != nil { return nil, err diff --git a/pkg/accountsdb/vote_index.go b/pkg/accountsdb/vote_index.go new file mode 100644 index 000000000..982fd8f62 --- /dev/null +++ b/pkg/accountsdb/vote_index.go @@ -0,0 +1,153 @@ +package accountsdb + +import ( + "context" + "fmt" + + "github.com/Overclock-Validator/mithril/pkg/addresses" + "github.com/cockroachdb/pebble" + "github.com/gagliardetto/solana-go" +) + +// Candidate keys are separate from the 32-byte account-index keys. They are +// append-only; callers check the current account state before returning it. +const voteIndexPrefix = "\x00mithril.vote." + +var voteIndexReadyKey = []byte("\x00mithril.meta.vote_index_ready") + +func voteIndexKey(pubkey solana.PublicKey) []byte { + key := make([]byte, len(voteIndexPrefix)+len(pubkey)) + copy(key, voteIndexPrefix) + copy(key[len(voteIndexPrefix):], pubkey[:]) + return key +} + +// SeedVoteAccountPubkeys seeds a freshly built snapshot index with vote-program +// account candidates observed while parsing its appendvecs. +func (db *AccountsDb) SeedVoteAccountPubkeys(pubkeys []solana.PublicKey) error { + batch := db.Index.NewBatch() + defer batch.Close() + for _, pubkey := range pubkeys { + if err := batch.Set(voteIndexKey(pubkey), nil, nil); err != nil { + return err + } + } + if err := batch.Set(voteIndexReadyKey, nil, nil); err != nil { + return err + } + if err := batch.Commit(pebble.Sync); err != nil { + return err + } + if db.IndexWALDisabled { + return db.Index.Flush() + } + return nil +} + +// VoteAccountPubkeys returns every known vote-account candidate. Existing +// stores built before this index was added are migrated once, on first use. +func (db *AccountsDb) VoteAccountPubkeys(ctx context.Context) ([]solana.PublicKey, error) { + db.voteIndexMu.Lock() + defer db.voteIndexMu.Unlock() + if _, closer, err := db.Index.Get(voteIndexReadyKey); err == nil { + if err := closer.Close(); err != nil { + return nil, err + } + } else if err == pebble.ErrNotFound { + if err := db.migrateVoteIndex(ctx); err != nil { + return nil, err + } + } else { + return nil, err + } + + iter, err := db.Index.NewIter(&pebble.IterOptions{ + LowerBound: []byte(voteIndexPrefix), + UpperBound: []byte("\x00mithril.vote/"), + }) + if err != nil { + return nil, err + } + var pubkeys []solana.PublicKey + for iter.First(); iter.Valid(); iter.Next() { + if err := ctx.Err(); err != nil { + iter.Close() + return nil, err + } + key := iter.Key() + if len(key) != len(voteIndexPrefix)+32 { + iter.Close() + return nil, fmt.Errorf("invalid vote-index key length %d", len(key)) + } + pubkeys = append(pubkeys, solana.PublicKeyFromBytes(key[len(voteIndexPrefix):])) + } + iterErr := iter.Error() + closeErr := iter.Close() + if iterErr != nil { + return nil, iterErr + } + return pubkeys, closeErr +} + +// migrateVoteIndex scans an older store once. Finalized folds can run during +// the scan: they add their new vote candidates to the same append-only index. +func (db *AccountsDb) migrateVoteIndex(ctx context.Context) error { + iter, err := db.Index.NewIter(nil) + if err != nil { + return err + } + defer iter.Close() + batch := db.Index.NewBatch() + defer batch.Close() + keys := make([]solana.PublicKey, 0, 512) + flush := func() error { + if len(keys) == 0 { + return nil + } + accounts, err := db.GetAccountsBatch(ctx, db.DurableThrough(), keys) + if err != nil { + return err + } + for i, account := range accounts { + if account != nil && account.Lamports > 0 && account.Owner == addresses.VoteProgramAddr { + if err := batch.Set(voteIndexKey(keys[i]), nil, nil); err != nil { + return err + } + } + } + keys = keys[:0] + return nil + } + for iter.First(); iter.Valid(); iter.Next() { + if err := ctx.Err(); err != nil { + return err + } + if key := iter.Key(); len(key) == 32 { + keys = append(keys, solana.PublicKeyFromBytes(key)) + if len(keys) == cap(keys) { + if err := flush(); err != nil { + return err + } + } + } + } + if err := iter.Error(); err != nil { + return err + } + if err := flush(); err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + if err := batch.Set(voteIndexReadyKey, nil, nil); err != nil { + return err + } + if err := batch.Commit(pebble.Sync); err != nil { + return err + } + if db.IndexWALDisabled { + return db.Index.Flush() + } + return nil +} diff --git a/pkg/rpcserver/get_vote_accounts.go b/pkg/rpcserver/get_vote_accounts.go index b6eddddfe..1025fef08 100644 --- a/pkg/rpcserver/get_vote_accounts.go +++ b/pkg/rpcserver/get_vote_accounts.go @@ -7,6 +7,7 @@ import ( "sync" "github.com/Overclock-Validator/mithril/pkg/addresses" + "github.com/Overclock-Validator/mithril/pkg/features" "github.com/Overclock-Validator/mithril/pkg/global" "github.com/Overclock-Validator/mithril/pkg/sealevel" "github.com/filecoin-project/go-jsonrpc" @@ -75,8 +76,11 @@ func (rpcServer *RpcServer) GetVoteAccounts(ctx context.Context, p jsonrpc.RawPa return GetVoteAccountsResp{}, err } - votePubkeys := rpcServer.acctsDb.VoteAccountPubkeys() - // A fresh snapshot can have an empty vote cache before any vote account is read. + votePubkeys, err := rpcServer.acctsDb.VoteAccountPubkeys(ctx) + if err != nil { + return GetVoteAccountsResp{}, fmt.Errorf("list vote accounts: %w", err) + } + // Epoch stakes also cover stores created before the durable index was built. for votePubkey := range stakes.Stakes { votePubkeys = append(votePubkeys, votePubkey) } @@ -105,12 +109,12 @@ func (rpcServer *RpcServer) GetVoteAccounts(ctx context.Context, p jsonrpc.RawPa } for index, votePubkey := range votePubkeys { account := accounts[index] - if account == nil || account.Owner != addresses.VoteProgramAddr { - return GetVoteAccountsResp{}, fmt.Errorf("vote account %s is unavailable at rooted slot %d", votePubkey, rooted.Slot) + if account == nil || account.Lamports == 0 || account.Owner != addresses.VoteProgramAddr { + continue // An old candidate may have been deleted or changed owner. } versioned, err := sealevel.UnmarshalVersionedVoteState(account.Data) - if err != nil { - return GetVoteAccountsResp{}, fmt.Errorf("decode vote account %s: %w", votePubkey, err) + if err != nil || !versioned.IsInitialized() { + continue // Exclude invalid and uninitialized vote states. } voteState := versioned.ConvertToCurrent() lastVote, _ := voteState.LastVotedSlot() @@ -152,7 +156,12 @@ func (rpcServer *RpcServer) GetVoteAccounts(ctx context.Context, p jsonrpc.RawPa } func (rpcServer *RpcServer) rootedActivatedStakes(ctx context.Context, slot, epoch uint64) (map[solana.PublicKey]uint64, error) { - rooted, accounts, err := rpcServer.readRootedAccounts(ctx, []solana.PublicKey{sealevel.SysvarStakeHistoryAddr}) + // Replay may not have published a slot context after snapshot boot; both + // inputs to stake activation must come from the same rooted bank instead. + rooted, accounts, err := rpcServer.readRootedAccounts(ctx, []solana.PublicKey{ + sealevel.SysvarStakeHistoryAddr, + features.ReduceStakeWarmupCooldown.Address, + }) if err != nil { return nil, fmt.Errorf("read rooted stake history: %w", err) } @@ -166,11 +175,18 @@ func (rpcServer *RpcServer) rootedActivatedStakes(ctx context.Context, slot, epo if err := history.UnmarshalWithDecoder(bin.NewBinDecoder(accounts[0].Data)); err != nil { return nil, fmt.Errorf("decode stake history at rooted slot %d: %w", slot, err) } - slotCtx := rpcServer.getSlotCtx() - if slotCtx == nil || slotCtx.Features == nil { - return nil, fmt.Errorf("node features unavailable for stake activation") + var activationEpoch *uint64 + featureAccount := accounts[1] + if featureAccount != nil && featureAccount.Lamports > 0 && featureAccount.Owner == addresses.FeatureAddr { + var feature features.FeatureAcct + if err := feature.UnmarshalWithDecoder(bin.NewBinDecoder(featureAccount.Data)); err != nil { + return nil, fmt.Errorf("decode rooted stake warmup feature: %w", err) + } + if feature.ActivatedAt != nil && *feature.ActivatedAt <= rooted.Slot { + epoch := rpcServer.epochSchedule.GetEpoch(*feature.ActivatedAt) + activationEpoch = &epoch + } } - activationEpoch := sealevel.NewWarmupCooldownRateEpochWithSlotCtx(slotCtx, rpcServer.epochSchedule) stakes := make(map[solana.PublicKey]uint64) var mu sync.Mutex _, err = global.StreamStakeAccounts(rpcServer.acctsDb, slot, func(_ solana.PublicKey, delegation *sealevel.Delegation, _ uint64) { diff --git a/pkg/snapshot/build_db.go b/pkg/snapshot/build_db.go index a7e2ed934..4853b2d78 100644 --- a/pkg/snapshot/build_db.go +++ b/pkg/snapshot/build_db.go @@ -19,6 +19,7 @@ import ( "github.com/Overclock-Validator/mithril/pkg/statsd" "github.com/Overclock-Validator/mithril/pkg/txstatus" "github.com/cockroachdb/pebble" + "github.com/gagliardetto/solana-go" "github.com/panjf2000/ants/v2" ) @@ -309,12 +310,12 @@ func BuildAccountsDbPaths( numShards := snapshotIndexShards() sl := NewShardLogger(numShards, logsDir) - // Create stake pubkey collector for building stake index during appendvec processing - stakeCollector := &stakeIndexCollector{ - entries: make([]accountsdb.StakeIndexEntry, 0, 1000000), // Pre-allocate for ~1M stake accounts + // Collect stake and vote pubkeys while parsing appendvecs. + accountCollector := &snapshotAccountCollector{ + stakeEntries: make([]accountsdb.StakeIndexEntry, 0, 1000000), // Pre-allocate for ~1M stake accounts } - pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, accountsDbDir, &largestFileId, stakeCollector) + pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, accountsDbDir, &largestFileId, accountCollector) if err != nil { return nil, nil, fmt.Errorf("initializing worker pools: %w", err) } @@ -416,7 +417,7 @@ func BuildAccountsDbPaths( // Write stake pubkey index file (with appendvec location hints) stakeIndexPath := filepath.Join(accountsDbDir, "stake_pubkeys.idx") - if err := accountsdb.WriteStakePubkeyIndex(stakeIndexPath, stakeCollector.entries); err != nil { + if err := accountsdb.WriteStakePubkeyIndex(stakeIndexPath, accountCollector.stakeEntries); err != nil { return nil, nil, fmt.Errorf("writing stake pubkey index: %w", err) } @@ -431,6 +432,10 @@ func BuildAccountsDbPaths( if err != nil { return nil, nil, err } + if err := accountsDb.SeedVoteAccountPubkeys(accountCollector.votePubkeys); err != nil { + accountsDb.CloseDb() + return nil, nil, fmt.Errorf("seeding vote pubkey index: %w", err) + } if incrementalManifest != nil { return accountsDb, incrementalManifest, nil @@ -646,29 +651,22 @@ func invokeSnapshotTask(wg *sync.WaitGroup, pool *ants.PoolWithFunc, task any) ( return err } -// stakeIndexCollector aggregates stake account pubkeys from multiple worker goroutines -// during appendvec processing. Used to build the stake pubkey index file. -// -// WHY: The manifest's delegation list can be stale/incomplete (Firedancer notes: -// "the cache in the manifest is partially incomplete"). Instead of trusting manifest -// data, we: -// 1. Collect stake pubkeys during appendvec parsing (by checking owner == StakeProgramAddr) -// 2. Write them to stake_pubkeys.idx after snapshot processing -// 3. At startup, load pubkeys from index and read ALL delegation fields from AccountsDB -// -// This ensures stake cache contains fresh data from AccountsDB, not potentially stale -// manifest data. -type stakeIndexCollector struct { - mu sync.Mutex - entries []accountsdb.StakeIndexEntry +// snapshotAccountCollector records stake and vote pubkeys while workers parse +// appendvecs. The manifest delegation list can be incomplete, and vote accounts +// without stake are absent from it. +type snapshotAccountCollector struct { + mu sync.Mutex + stakeEntries []accountsdb.StakeIndexEntry + votePubkeys []solana.PublicKey } -func (c *stakeIndexCollector) Add(entries []accountsdb.StakeIndexEntry) { - if len(entries) == 0 { +func (c *snapshotAccountCollector) Add(stakes []accountsdb.StakeIndexEntry, votes []solana.PublicKey) { + if len(stakes) == 0 && len(votes) == 0 { return } c.mu.Lock() - c.entries = append(c.entries, entries...) + c.stakeEntries = append(c.stakeEntries, stakes...) + c.votePubkeys = append(c.votePubkeys, votes...) c.mu.Unlock() } @@ -679,7 +677,7 @@ func initWorkerPools( incrementalManifest *SnapshotManifest, accountsDbDir string, largestFileId *atomic.Uint64, - stakeCollector *stakeIndexCollector, + accountCollector *snapshotAccountCollector, ) (*snapshotWorkerPools, error) { indexEntryCommitterWorkers := snapshotIndexEntryCommitterWorkers() indexEntryBuilderWorkers := snapshotIndexEntryBuilderWorkers() @@ -722,14 +720,14 @@ func initWorkerPools( return } task := i.(indexEntryBuilderTask) - pubkeys, entries, stakeEntries, err := accountsdb.BuildIndexEntriesFromAppendVecs(task.Data, task.FileSize, task.Slot, task.FileId) + pubkeys, entries, stakeEntries, votePubkeys, err := accountsdb.BuildIndexEntriesFromAppendVecs(task.Data, task.FileSize, task.Slot, task.FileId) if err != nil { workerErrors.Record(fmt.Errorf("building index entries: %w", err)) return } - // Collect stake entries with appendvec location hints for building stake index - stakeCollector.Add(stakeEntries) + // Collect stake locations and vote pubkeys for their indexes. + accountCollector.Add(stakeEntries, votePubkeys) commitTask := indexEntryCommitterTask{IndexEntries: entries, Pubkeys: pubkeys} statsd.Timing(statsd.TasksIndexEntryBuilderLatency, uint64(time.Since(start)), nil) diff --git a/pkg/snapshot/build_db_with_incr.go b/pkg/snapshot/build_db_with_incr.go index 7afb8b9bf..9b737b7cf 100644 --- a/pkg/snapshot/build_db_with_incr.go +++ b/pkg/snapshot/build_db_with_incr.go @@ -81,12 +81,12 @@ func BuildAccountsDbAuto( defer cleanupIndexWorkDir() sl := NewShardLogger(numShards, logsDir) - // Create stake pubkey collector for building stake index during appendvec processing - stakeCollector := &stakeIndexCollector{ - entries: make([]accountsdb.StakeIndexEntry, 0, 1000000), // Pre-allocate for ~1M stake accounts + // Collect stake and vote pubkeys while parsing appendvecs. + accountCollector := &snapshotAccountCollector{ + stakeEntries: make([]accountsdb.StakeIndexEntry, 0, 1000000), // Pre-allocate for ~1M stake accounts } - pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, accountsDbDir, &largestFileId, stakeCollector) + pools, err := initWorkerPools(wg, sl, manifest, incrementalManifest, accountsDbDir, &largestFileId, accountCollector) if err != nil { return nil, nil, fmt.Errorf("initializing worker pools: %w", err) } @@ -287,7 +287,7 @@ func BuildAccountsDbAuto( // Write stake pubkey index file (with appendvec location hints) stakeIndexPath := filepath.Join(accountsDbDir, "stake_pubkeys.idx") - if err := accountsdb.WriteStakePubkeyIndex(stakeIndexPath, stakeCollector.entries); err != nil { + if err := accountsdb.WriteStakePubkeyIndex(stakeIndexPath, accountCollector.stakeEntries); err != nil { return nil, nil, fmt.Errorf("writing stake pubkey index: %w", err) } @@ -302,6 +302,10 @@ func BuildAccountsDbAuto( if err != nil { return nil, nil, err } + if err := accountsDb.SeedVoteAccountPubkeys(accountCollector.votePubkeys); err != nil { + accountsDb.CloseDb() + return nil, nil, fmt.Errorf("seeding vote pubkey index: %w", err) + } rpcClient := rpcclient.NewRpcClient(rpcEndpoints[0]) latestSlot, err := rpcClient.GetSlot() From f09f3574e84927228074a1a0329fcd307d8f1add Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Thu, 24 Sep 2026 13:11:10 +0530 Subject: [PATCH 4/6] test(rpc): cover vote index recovery and snapshot-start metrics --- pkg/accountsdb/appendvec_test.go | 24 ++++++--- pkg/accountsdb/fold_test.go | 2 +- pkg/accountsdb/vote_index_test.go | 70 +++++++++++++++++++++++++ pkg/rpcserver/get_vote_accounts_test.go | 46 ++++++++++++++-- pkg/snapshot/build_db_worker_test.go | 2 +- 5 files changed, 133 insertions(+), 11 deletions(-) create mode 100644 pkg/accountsdb/vote_index_test.go diff --git a/pkg/accountsdb/appendvec_test.go b/pkg/accountsdb/appendvec_test.go index 9314740a1..09c6c4a39 100644 --- a/pkg/accountsdb/appendvec_test.go +++ b/pkg/accountsdb/appendvec_test.go @@ -5,6 +5,7 @@ import ( "encoding/binary" "testing" + "github.com/Overclock-Validator/mithril/pkg/addresses" "github.com/gagliardetto/solana-go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -21,7 +22,7 @@ func TestBuildIndexEntriesDoesNotReadPastSliceLen(t *testing.T) { copy(backing[len(real):], phantom) data := backing[:len(real)] // len excludes phantom; cap and manifest include it. - pubkeys, entries, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(cap(data)), 7, 9) + pubkeys, entries, _, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(cap(data)), 7, 9) require.NoError(t, err) require.Equal(t, []solana.PublicKey{realKey}, pubkeys) require.Len(t, entries, 1) @@ -37,7 +38,7 @@ func TestBuildIndexEntriesBoundsOversizedManifestByDataLen(t *testing.T) { var pubkeys []solana.PublicKey require.NotPanics(t, func() { var err error - pubkeys, _, _, err = BuildIndexEntriesFromAppendVecs(data, uint64(len(data)+hdrLen), 8, 10) + pubkeys, _, _, _, err = BuildIndexEntriesFromAppendVecs(data, uint64(len(data)+hdrLen), 8, 10) require.NoError(t, err) }) require.Equal(t, []solana.PublicKey{key}, pubkeys) @@ -54,7 +55,7 @@ func TestBuildIndexEntriesStopsAtDefaultZeroLamportTerminator(t *testing.T) { binary.LittleEndian.PutUint64(terminator[dataLenOffset:dataLenOffset+8], ^uint64(0)) data := append(append(append([]byte{}, first...), terminator...), afterTerminator...) - pubkeys, entries, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 12, 13) + pubkeys, entries, _, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 12, 13) require.NoError(t, err) require.Equal(t, []solana.PublicKey{firstKey}, pubkeys) require.Len(t, entries, 1) @@ -67,7 +68,7 @@ func TestBuildIndexEntriesTerminatorRequiresDefaultKeyAndZeroLamports(t *testing zeroLamportAccount := marshalAppendVecTestAccount(t, zeroLamportKey, 0, []byte("tombstone")) data := append(defaultKeyAccount, zeroLamportAccount...) - pubkeys, entries, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 18, 19) + pubkeys, entries, _, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 18, 19) require.NoError(t, err) require.Equal(t, []solana.PublicKey{defaultKey, zeroLamportKey}, pubkeys) require.Len(t, entries, 2) @@ -80,7 +81,7 @@ func TestBuildIndexEntriesRejectsTruncatedAccountData(t *testing.T) { copy(data[pubkeyOffset:pubkeyOffset+32], key[:]) binary.LittleEndian.PutUint64(data[lamportsOffset:lamportsOffset+8], 1) - pubkeys, entries, stakeEntries, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 14, 15) + pubkeys, entries, stakeEntries, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 14, 15) require.ErrorContains(t, err, "truncated appendvec account data") assert.Nil(t, pubkeys) assert.Nil(t, entries) @@ -92,12 +93,23 @@ func TestBuildIndexEntriesAcceptsFinalAccountWithoutAlignmentPadding(t *testing. encoded := marshalAppendVecTestAccount(t, key, 77, []byte{1, 2, 3}) data := encoded[: hdrLen+3 : hdrLen+3] - pubkeys, entries, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(encoded)), 16, 17) + pubkeys, entries, _, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(encoded)), 16, 17) require.NoError(t, err) require.Equal(t, []solana.PublicKey{key}, pubkeys) require.Len(t, entries, 1) } +func TestBuildIndexEntriesCollectsVoteAccounts(t *testing.T) { + vote := appendVecTestPubkey(10) + ordinary := appendVecTestPubkey(11) + voteData := marshalAppendVecTestAccount(t, vote, 1, nil) + copy(voteData[ownerOffset:ownerOffset+32], addresses.VoteProgramAddr[:]) + data := append(voteData, marshalAppendVecTestAccount(t, ordinary, 1, nil)...) + _, _, _, votes, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 20, 21) + require.NoError(t, err) + require.Equal(t, []solana.PublicKey{vote}, votes) +} + func appendVecTestPubkey(firstByte byte) solana.PublicKey { var key solana.PublicKey key[0] = firstByte diff --git a/pkg/accountsdb/fold_test.go b/pkg/accountsdb/fold_test.go index 64f01d0f4..3f9c24033 100644 --- a/pkg/accountsdb/fold_test.go +++ b/pkg/accountsdb/fold_test.go @@ -144,7 +144,7 @@ func TestSegmentManifestRoundTripAndTornDetection(t *testing.T) { Bankhashes: []SlotBankhash{{Slot: 100, Bankhash: bh(100)}, {Slot: 130, Bankhash: bh(130)}}, Records: []ManifestRecord{ {Pubkey: [32]byte{1}, Offset: 0, OwnerSlot: 100, PrevValid: true, Prev: AccountIndexEntry{Slot: 90, FileId: 3, Offset: 77}}, - {Pubkey: [32]byte{2}, Offset: 136, OwnerSlot: 130}, + {Pubkey: [32]byte{2}, Offset: 136, OwnerSlot: 130, Vote: true}, }, ResumeCtx: []byte(`{"slot":130}`), } diff --git a/pkg/accountsdb/vote_index_test.go b/pkg/accountsdb/vote_index_test.go new file mode 100644 index 000000000..bdbbc3006 --- /dev/null +++ b/pkg/accountsdb/vote_index_test.go @@ -0,0 +1,70 @@ +package accountsdb + +import ( + "context" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + "github.com/Overclock-Validator/mithril/pkg/addresses" + "github.com/cockroachdb/pebble" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestVoteIndexMigratesExistingStoreAndPersists(t *testing.T) { + db, dir := newFoldTestDb(t) + vote := &accounts.Account{Key: solana.PublicKey{1}, Owner: addresses.VoteProgramAddr, Lamports: 1} + ordinary := &accounts.Account{Key: solana.PublicKey{2}, Lamports: 1} + _, err := db.CommitBatch(foldDeltas(accounts.SlotDelta{Slot: 10, Delta: []*accounts.Account{vote, ordinary}}), 10, nil, nil) + require.NoError(t, err) + // Mimic a database written before the vote index existed. + require.NoError(t, db.Index.Delete(voteIndexKey(vote.Key), pebble.Sync)) + db.VoteAcctCache.Clear() + got, err := db.VoteAccountPubkeys(context.Background()) + require.NoError(t, err) + require.Equal(t, []solana.PublicKey{vote.Key}, got) + + db = reopenFoldTestDb(t, db, dir) + defer db.CloseDb() + _, err = db.RecoverFoldState() + require.NoError(t, err) + got, err = db.VoteAccountPubkeys(context.Background()) + require.NoError(t, err) + require.Equal(t, []solana.PublicKey{vote.Key}, got) +} + +func TestVoteIndexSnapshotSeedAndFoldRecovery(t *testing.T) { + db, dir := newFoldTestDb(t) + seedVote := solana.PublicKey{1} + foldVote := &accounts.Account{Key: solana.PublicKey{2}, Owner: addresses.VoteProgramAddr, Lamports: 1} + require.NoError(t, db.SeedVoteAccountPubkeys([]solana.PublicKey{seedVote})) + _, err := db.CommitBatch(foldDeltas(accounts.SlotDelta{Slot: 10, Delta: []*accounts.Account{foldVote}}), 10, nil, nil) + require.NoError(t, err) + got, err := db.VoteAccountPubkeys(context.Background()) + require.NoError(t, err) + require.Equal(t, []solana.PublicKey{seedVote, foldVote.Key}, got) + + // Simulate a lost index tail: recovery must restore the candidate from the + // durable fold manifest along with the account's primary index entry. + require.NoError(t, db.Index.Delete(voteIndexKey(foldVote.Key), pebble.Sync)) + require.NoError(t, db.Index.Delete(metaKeyLastBatch, pebble.Sync)) + db = reopenFoldTestDb(t, db, dir) + defer db.CloseDb() + _, err = db.RecoverFoldState() + require.NoError(t, err) + got, err = db.VoteAccountPubkeys(context.Background()) + require.NoError(t, err) + require.Equal(t, []solana.PublicKey{seedVote, foldVote.Key}, got) +} + +func TestVoteIndexMigrationCanRetryAfterCancellation(t *testing.T) { + db, _ := newFoldTestDb(t) + defer db.CloseDb() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := db.VoteAccountPubkeys(ctx) + require.ErrorIs(t, err, context.Canceled) + got, err := db.VoteAccountPubkeys(context.Background()) + require.NoError(t, err) + require.Empty(t, got) +} diff --git a/pkg/rpcserver/get_vote_accounts_test.go b/pkg/rpcserver/get_vote_accounts_test.go index d6f9d35c3..b8671f449 100644 --- a/pkg/rpcserver/get_vote_accounts_test.go +++ b/pkg/rpcserver/get_vote_accounts_test.go @@ -26,6 +26,7 @@ func TestGetVoteAccountsUsesRootedVoteState(t *testing.T) { currentVote := solana.PublicKey{1} delinquentVote := solana.PublicKey{2} newVote := solana.PublicKey{3} + invalidVote := solana.PublicKey{5} currentNode := solana.PublicKey{11} delinquentNode := solana.PublicKey{12} newNode := solana.PublicKey{13} @@ -34,6 +35,7 @@ func TestGetVoteAccountsUsesRootedVoteState(t *testing.T) { currentState := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionV4} currentState.V4.NodePubkey = currentNode + currentState.V4.AuthorizedVoters.AuthorizedVoters.Set(0, currentNode) currentState.V4.InflationRewardsCommissionBps = 755 currentState.V4.RootSlot = ¤tRoot currentState.V4.Votes.PushBack(sealevel.LandedVote{Lockout: sealevel.VoteLockout{Slot: 499}}) @@ -48,13 +50,17 @@ func TestGetVoteAccountsUsesRootedVoteState(t *testing.T) { delinquentState := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionCurrent} delinquentState.Current.NodePubkey = delinquentNode + delinquentState.Current.AuthorizedVoters.AuthorizedVoters.Set(0, delinquentNode) delinquentState.Current.Commission = 5 delinquentState.Current.RootSlot = &delinquentRoot delinquentState.Current.Votes.PushBack(sealevel.LandedVote{Lockout: sealevel.VoteLockout{Slot: 300}}) newState := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionCurrent} newState.Current.NodePubkey = newNode + newState.Current.AuthorizedVoters.AuthorizedVoters.Set(0, newNode) newState.Current.Votes.PushBack(sealevel.LandedVote{Lockout: sealevel.VoteLockout{Slot: 498}}) + uninitializedVote := solana.PublicKey{4} + uninitializedState := sealevel.VoteStateVersions{Type: sealevel.VoteStateVersionCurrent} db := newRPCAccountsDB(t) global.ClearPendingStakePubkeys() @@ -84,9 +90,11 @@ func TestGetVoteAccountsUsesRootedVoteState(t *testing.T) { voteAccountForRPC(t, currentVote, ¤tState), voteAccountForRPC(t, delinquentVote, &delinquentState), voteAccountForRPC(t, newVote, &newState), + voteAccountForRPC(t, uninitializedVote, &uninitializedState), + {Key: invalidVote, Owner: addresses.VoteProgramAddr, Lamports: 1}, stakeAccount(solana.PublicKey{21}, currentVote, 125), stakeAccount(solana.PublicKey{22}, delinquentVote, 200), - stakeAccount(solana.PublicKey{23}, solana.PublicKey{4}, 50), + stakeAccount(solana.PublicKey{23}, uninitializedVote, 50), {Key: sealevel.SysvarStakeHistoryAddr, Owner: addresses.SysvarOwnerAddr, Lamports: 1, Data: historyData.Bytes()}, }, }}, rootedSlot, nil, nil) @@ -117,7 +125,6 @@ func TestGetVoteAccountsUsesRootedVoteState(t *testing.T) { SlotsPerEpoch: 1_000, }, } - server.SetSlotCtx(&sealevel.SlotCtx{Features: features.NewFeaturesDefault()}) server.SetRootedBankState(rootedSlot, 490, 1_000) got, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{ map[string]interface{}{"commitment": "confirmed"}, @@ -149,10 +156,43 @@ func TestGetVoteAccountsUsesRootedVoteState(t *testing.T) { db.VoteAcctCache.Clear() cold, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{})) require.NoError(t, err) - require.Len(t, cold.Current, 1) + require.Len(t, cold.Current, 2) require.Equal(t, currentVote.String(), cold.Current[0].VotePubkey) + require.Equal(t, newVote.String(), cold.Current[1].VotePubkey) require.Len(t, cold.Delinquent, 1) require.Equal(t, delinquentVote.String(), cold.Delinquent[0].VotePubkey) + + // Candidate keys are append-only, but an account that changes owner must + // no longer be returned as a vote account. + _, err = db.CommitBatch([]accounts.SlotDelta{{Slot: 501, Delta: []*accounts.Account{{ + Key: newVote, Lamports: 1, Owner: addresses.SystemProgramAddr, + }}}}, 501, nil, nil) + require.NoError(t, err) + server.SetRootedBankState(501, 490, 1_000) + changed, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{})) + require.NoError(t, err) + require.Len(t, changed.Current, 1) + require.Equal(t, currentVote.String(), changed.Current[0].VotePubkey) + + activationSlot := uint64(0) + featureData, err := features.MarshalFeatureAcct(&features.FeatureAcct{ActivatedAt: &activationSlot}) + require.NoError(t, err) + _, err = db.CommitBatch([]accounts.SlotDelta{{Slot: 502, Delta: []*accounts.Account{{ + Key: features.ReduceStakeWarmupCooldown.Address, Owner: addresses.FeatureAddr, Lamports: 1, Data: featureData, + }}}}, 502, nil, nil) + require.NoError(t, err) + server.SetRootedBankState(502, 490, 1_000) + withFeature, err := server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{})) + require.NoError(t, err) + require.Equal(t, uint64(125), withFeature.Current[0].ActivatedStake) + + _, err = db.CommitBatch([]accounts.SlotDelta{{Slot: 503, Delta: []*accounts.Account{{ + Key: features.ReduceStakeWarmupCooldown.Address, Owner: addresses.FeatureAddr, Lamports: 1, Data: []byte{1}, + }}}}, 503, nil, nil) + require.NoError(t, err) + server.SetRootedBankState(503, 490, 1_000) + _, err = server.GetVoteAccounts(t.Context(), mustRawParams(t, []interface{}{})) + require.ErrorContains(t, err, "decode rooted stake warmup feature") } func TestGetVoteAccountsRejectsInvalidConfig(t *testing.T) { diff --git a/pkg/snapshot/build_db_worker_test.go b/pkg/snapshot/build_db_worker_test.go index a8ab894d6..9d36fc3c9 100644 --- a/pkg/snapshot/build_db_worker_test.go +++ b/pkg/snapshot/build_db_worker_test.go @@ -59,7 +59,7 @@ func TestSnapshotWorkerParseFailurePropagatesAfterDrain(t *testing.T) { nil, accountsDir, &atomic.Uint64{}, - &stakeIndexCollector{}, + &snapshotAccountCollector{}, ) require.NoError(t, err) t.Cleanup(pools.Release) From 5651a73782cda1c7f4a5781bc21718bae8b9aef0 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Sat, 26 Sep 2026 15:15:24 +0530 Subject: [PATCH 5/6] Preserve replay state setter name for MCP integration --- cmd/mithril/node/node.go | 4 ++-- pkg/replay/block.go | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index f9871e85f..cf992c9cf 100644 --- a/cmd/mithril/node/node.go +++ b/cmd/mithril/node/node.go @@ -2932,7 +2932,7 @@ postBootstrap: } } - var slotCtxSetter replay.RPCStateSetter + var slotCtxSetter replay.SlotCtxSetter if rpcServer != nil { slotCtxSetter = rpcServer } @@ -4405,7 +4405,7 @@ func runReplayWithRecovery( useTurbine bool, dbgOpts *replay.DebugOptions, metricsWriter io.Writer, - rpcServer replay.RPCStateSetter, + rpcServer replay.SlotCtxSetter, mithrilState *state.MithrilState, blockFetchOpts *replay.BlockFetchOpts, consensusOpts *replay.ConsensusOpts, diff --git a/pkg/replay/block.go b/pkg/replay/block.go index 5c962a264..a3b7ae2ad 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -47,9 +47,9 @@ import ( "github.com/panjf2000/ants/v2" ) -// RPCStateSetter is implemented by the RPC server so replay can publish both +// SlotCtxSetter is implemented by the RPC server so replay can publish both // its live execution bank and the latest bank durably folded into AccountsDB. -type RPCStateSetter interface { +type SlotCtxSetter interface { SetSlotCtx(slotCtx *sealevel.SlotCtx) SetRootedBankState(slot, blockHeight, transactionCount uint64) } @@ -1684,7 +1684,7 @@ func ReplayBlocks( useTurbine bool, dbgOpts *DebugOptions, metricsWriter io.Writer, - rpcServer RPCStateSetter, + rpcServer SlotCtxSetter, blockFetchOpts *BlockFetchOpts, consensusOpts *ConsensusOpts, // nil = use defaults (max_depth=64, policy="halt") onCancelWriteState OnCancelWriteState, // callback to write state immediately on cancellation (can be nil) From ab30fded26c7dc0645776d7263f717faa9b00d9f Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Sun, 27 Sep 2026 19:24:05 +0530 Subject: [PATCH 6/6] rpc: add commitment-aware getSlot with minimum context validation Serve the live replay slot for processed requests and the published rooted slot for confirmed and finalized requests. Decode minContextSlot as an exact uint64 and return the existing structured RPC errors. Add HTTP coverage for commitment selection, minimum-context boundaries, invalid inputs, numeric precision, missing roots and cancellation. Document the supported options. --- README.md | 6 ++ pkg/rpcserver/get_slot.go | 56 +++++++++++++++ pkg/rpcserver/get_slot_test.go | 121 ++++++++++++++++++++++++++++++++ pkg/rpcserver/rpcserver.go | 1 + pkg/rpcserver/rpcserver_test.go | 4 +- 5 files changed, 186 insertions(+), 2 deletions(-) create mode 100644 pkg/rpcserver/get_slot.go create mode 100644 pkg/rpcserver/get_slot_test.go diff --git a/README.md b/README.md index dcbb27af2..dc143441a 100644 --- a/README.md +++ b/README.md @@ -223,6 +223,12 @@ curl http://YOUR_MITHRIL_IP:8899 -X POST -H "Content-Type: application/json" -d - `getBlockHeight` - Get current block height - `getEpochInfo` - Get current epoch info - `getLatestBlockhash` - Get recent blockhash +- `getSlot` - Get the slot at the requested commitment (defaults to finalized) + +`getSlot` accepts an optional config object with `commitment` and `minContextSlot`. +On Alpenglow, `processed` uses the live replay slot; `confirmed` and `finalized` +use the published rooted slot. If that slot is below `minContextSlot`, the request +returns error `-32016` with the available slot in `data.contextSlot`. We're actively expanding RPC method coverage. Upcoming methods include transaction simulation, send transaction, and get leader schedule. diff --git a/pkg/rpcserver/get_slot.go b/pkg/rpcserver/get_slot.go new file mode 100644 index 000000000..7dbb52fba --- /dev/null +++ b/pkg/rpcserver/get_slot.go @@ -0,0 +1,56 @@ +package rpcserver + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/filecoin-project/go-jsonrpc" +) + +// GetSlot returns the slot at the requested commitment, defaulting to finalized. +func (rpcServer *RpcServer) GetSlot(ctx context.Context, p jsonrpc.RawParams) (uint64, error) { + if err := ctx.Err(); err != nil { + return 0, err + } + var params []json.RawMessage + if len(p) != 0 { + if err := json.Unmarshal(p, ¶ms); err != nil { + return 0, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + } + if len(params) > 1 { + return 0, &InvalidParamsError{Message: "getSlot accepts at most one config object"} + } + var config struct { + Commitment *string `json:"commitment"` + MinContextSlot *uint64 `json:"minContextSlot"` + } + if len(params) == 1 { + if err := json.Unmarshal(params[0], &config); err != nil { + return 0, &InvalidParamsError{Message: fmt.Sprintf("invalid getSlot config: %v", err)} + } + } + commitment := "finalized" + if config.Commitment != nil { + commitment = *config.Commitment + } + if commitment != "processed" && commitment != "confirmed" && commitment != "finalized" { + return 0, &InvalidParamsError{Message: "invalid commitment"} + } + + slot := global.Slot() + if commitment != "processed" { + // Match getEpochInfo's published rooted view for confirmed and finalized. + rooted, ok := rpcServer.getRootedBankState() + if !ok { + return 0, fmt.Errorf("node has no rooted bank available") + } + slot = rooted.Slot + } + if config.MinContextSlot != nil && slot < *config.MinContextSlot { + return 0, &MinContextSlotNotReachedError{ContextSlot: slot} + } + return slot, nil +} diff --git a/pkg/rpcserver/get_slot_test.go b/pkg/rpcserver/get_slot_test.go new file mode 100644 index 000000000..a61e7780f --- /dev/null +++ b/pkg/rpcserver/get_slot_test.go @@ -0,0 +1,121 @@ +package rpcserver + +import ( + "context" + "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/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestGetSlotHTTP(t *testing.T) { + oldSlot := global.Slot() + t.Cleanup(func() { global.SetSlot(oldSlot) }) + global.SetSlot(150) + server := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 512}, solana.Hash{}) + t.Cleanup(func() { require.NoError(t, server.listener.Close()) }) + server.SetRootedBankState(100, 90, 0) + httpServer := httptest.NewServer(server) + t.Cleanup(httpServer.Close) + + check := func(t *testing.T, params string, want uint64, code int) { + t.Helper() + body := `{"jsonrpc":"2.0","id":7,"method":"getSlot"` + if params != "" { + body += `,"params":` + params + } + resp, err := httpServer.Client().Post(httpServer.URL, "application/json", strings.NewReader(body+`}`)) + require.NoError(t, err) + defer resp.Body.Close() + var got struct { + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` + Result json.RawMessage `json:"result"` + Error *struct { + Code int `json:"code"` + Data struct { + ContextSlot uint64 `json:"contextSlot"` + } `json:"data"` + } `json:"error"` + } + require.NoError(t, json.NewDecoder(resp.Body).Decode(&got)) + require.Equal(t, "2.0", got.JSONRPC) + require.Equal(t, 7, got.ID) + if code != 0 { + require.NotNil(t, got.Error) + require.Equal(t, code, got.Error.Code) + require.Empty(t, got.Result) + if code == -32016 { + require.Equal(t, want, got.Error.Data.ContextSlot) + } + return + } + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Nil(t, got.Error) + require.Equal(t, fmt.Sprint(want), string(got.Result)) + } + + for _, tc := range []struct { + name string + params string + want uint64 + code int + }{ + {"omitted", "", 100, 0}, + {"empty", `[]`, 100, 0}, + {"null params", `null`, 100, 0}, + {"null config", `[null]`, 100, 0}, + {"empty config", `[{}]`, 100, 0}, + {"null fields", `[{"commitment":null,"minContextSlot":null}]`, 100, 0}, + {"finalized", `[{"commitment":"finalized"}]`, 100, 0}, + {"confirmed", `[{"commitment":"confirmed"}]`, 100, 0}, + {"processed", `[{"commitment":"processed"}]`, 150, 0}, + {"root boundary", `[{"minContextSlot":100}]`, 100, 0}, + {"root too old", `[{"minContextSlot":101}]`, 100, -32016}, + {"confirmed too old", `[{"commitment":"confirmed","minContextSlot":101}]`, 100, -32016}, + {"processed boundary", `[{"commitment":"processed","minContextSlot":150}]`, 150, 0}, + {"processed too old", `[{"commitment":"processed","minContextSlot":151}]`, 150, -32016}, + {"unknown field", `[{"futureOption":true}]`, 100, 0}, + {"extra argument", `[{},{}]`, 0, -32602}, + {"scalar config", `[true]`, 0, -32602}, + {"array config", `[[]]`, 0, -32602}, + {"unknown commitment", `[{"commitment":"invalid"}]`, 0, -32602}, + {"empty commitment", `[{"commitment":""}]`, 0, -32602}, + {"typed commitment", `[{"commitment":1}]`, 0, -32602}, + {"negative slot", `[{"minContextSlot":-1}]`, 0, -32602}, + {"fractional slot", `[{"minContextSlot":1.5}]`, 0, -32602}, + {"string slot", `[{"minContextSlot":"100"}]`, 0, -32602}, + {"overflow slot", `[{"minContextSlot":18446744073709551616}]`, 0, -32602}, + } { + t.Run(tc.name, func(t *testing.T) { check(t, tc.params, tc.want, tc.code) }) + } + + t.Run("exact uint64", func(t *testing.T) { + server.SetRootedBankState(1<<53, 0, 0) + check(t, `[{"minContextSlot":9007199254740993}]`, 1<<53, -32016) + server.SetRootedBankState(^uint64(0), 0, 0) + check(t, `[{"minContextSlot":18446744073709551615}]`, ^uint64(0), 0) + }) + t.Run("no rooted bank", func(t *testing.T) { + server.rootedBank.Store(nil) + check(t, `[]`, 0, 1) + check(t, `[{"commitment":"processed"}]`, 150, 0) + }) + t.Run("zero slot", func(t *testing.T) { + server.SetRootedBankState(0, 0, 0) + check(t, `[]`, 0, 0) + }) + t.Run("canceled request", func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err := server.GetSlot(ctx, nil) + require.ErrorIs(t, err, context.Canceled) + }) +} diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 2a50624a0..3f49e8886 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -69,6 +69,7 @@ var supportedRPCMethods = map[string]struct{}{ "getGenesisHash": {}, "getLatestBlockhash": {}, "getLeaderSchedule": {}, + "getSlot": {}, "getVoteAccounts": {}, "sendTransaction": {}, "simulateTransaction": {}, diff --git a/pkg/rpcserver/rpcserver_test.go b/pkg/rpcserver/rpcserver_test.go index 8ec2c4deb..7720037f7 100644 --- a/pkg/rpcserver/rpcserver_test.go +++ b/pkg/rpcserver/rpcserver_test.go @@ -10,7 +10,7 @@ import ( func TestServeHTTPQuietlyHandlesUnsupportedMethod(t *testing.T) { rpcServer := &RpcServer{} - req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"jsonrpc":"2.0","method":"getSlot","id":7}`)) + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"jsonrpc":"2.0","method":"unsupportedMethod","id":7}`)) rec := httptest.NewRecorder() rpcServer.ServeHTTP(rec, req) @@ -33,7 +33,7 @@ func TestServeHTTPQuietlyHandlesUnsupportedMethod(t *testing.T) { if resp.JSONRPC != "2.0" || resp.ID != 7 { t.Fatalf("unexpected response identity: %+v", resp) } - if resp.Error.Code != -32601 || resp.Error.Message != "method 'getSlot' not found" { + if resp.Error.Code != -32601 || resp.Error.Message != "method 'unsupportedMethod' not found" { t.Fatalf("unexpected error response: %+v", resp.Error) } }