From 03294f431d6700402a70e421668ebe1079ac01f5 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Wed, 23 Sep 2026 23:30:53 +0530 Subject: [PATCH 1/5] rpc: retain bounded rooted block metadata for Solana Exporter --- cmd/mithril/node/node.go | 11 + pkg/blockhistory/store.go | 353 +++++++++++++++++++++++++++++ pkg/replay/block.go | 49 +++- pkg/replay/block_fee_reward.go | 52 +++++ pkg/replay/leader_finalize.go | 3 + pkg/rpcserver/errors.go | 44 ++++ pkg/rpcserver/get_block_history.go | 150 ++++++++++++ pkg/rpcserver/rpcserver.go | 76 ++++++- pkg/snapshot/build_db.go | 1 + 9 files changed, 730 insertions(+), 9 deletions(-) create mode 100644 pkg/blockhistory/store.go create mode 100644 pkg/replay/block_fee_reward.go create mode 100644 pkg/rpcserver/get_block_history.go diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index c452d494e..eed8d0d1c 100644 --- a/cmd/mithril/node/node.go +++ b/cmd/mithril/node/node.go @@ -2517,6 +2517,10 @@ postBootstrap: klog.Fatalf("invalid port: %d", rpcPort) } else if rpcPort != 0 { rpcServer = rpcserver.NewRpcServer(accountsDb, uint16(rpcPort), epochScheduleFromState(mithrilState), solana.MustHashFromBase58(networkGenesisHash)) + retentionSlots := epochScheduleFromState(mithrilState).SlotsPerEpoch * 2 + if err := rpcServer.EnableBlockHistory(filepath.Join(accountsPath, "rpc-block-history"), retentionSlots, mithrilState.LastRootedSlot); err != nil { + klog.Fatalf("enable RPC block history: %v", err) + } rpcServer.Start() mlog.Log.Infof("Started RPC server on port %d", rpcPort) } @@ -4546,6 +4550,13 @@ func runReplayWithRecovery( break } } + if history, ok := rpcServer.(interface{ RewindBlockHistory(uint64) error }); ok { + if err := history.RewindBlockHistory(mithrilState.LastRootedSlot); err != nil { + result.Error = fmt.Errorf("fork switch: rewind RPC block history to slot %d: %w", mithrilState.LastRootedSlot, err) + mlog.Log.Errorf("%v; halting", result.Error) + break + } + } if mithrilState.LastRootedContext == nil { mlog.Log.Errorf("fork switch: no rooted checkpoint context to re-replay from; halting") break diff --git a/pkg/blockhistory/store.go b/pkg/blockhistory/store.go new file mode 100644 index 000000000..2d34fb56f --- /dev/null +++ b/pkg/blockhistory/store.go @@ -0,0 +1,353 @@ +package blockhistory + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/gagliardetto/solana-go" + "github.com/gagliardetto/solana-go/rpc" +) + +const recordVersion = 1 + +var ( + ErrNotAvailable = errors.New("block history is not available") + ErrSlotSkipped = errors.New("slot was skipped") +) + +type Record struct { + Slot uint64 `json:"slot"` + ParentSlot uint64 `json:"parentSlot,omitempty"` + BlockHeight uint64 `json:"blockHeight,omitempty"` + Blockhash string `json:"blockhash,omitempty"` + PreviousBlockhash string `json:"previousBlockhash,omitempty"` + BlockTime *int64 `json:"blockTime"` + Rewards []rpc.BlockReward `json:"rewards"` + Skipped bool `json:"skipped,omitempty"` +} + +type batch struct { + Version uint32 `json:"version"` + Through uint64 `json:"through"` + Records []Record `json:"records"` +} + +// Store retains speculative summaries in memory and writes one immutable batch +// immediately before the matching AccountsDB fold commits. Readers are also +// bounded by rooted, so an abandoned fork or interrupted fold stays invisible. +type Store struct { + dir string + retentionSlots uint64 + rooted atomic.Uint64 + mu sync.RWMutex + pending map[uint64]Record + persisted map[uint64]Record + sourceBatch map[uint64]uint64 + batches map[uint64][]uint64 +} + +func Open(dir string, retentionSlots uint64) (*Store, error) { + if retentionSlots == 0 { + return nil, errors.New("block history retention must be greater than zero") + } + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("create block history directory: %w", err) + } + store := &Store{ + dir: dir, + retentionSlots: retentionSlots, + pending: make(map[uint64]Record), + persisted: make(map[uint64]Record), + sourceBatch: make(map[uint64]uint64), + batches: make(map[uint64][]uint64), + } + entries, err := os.ReadDir(dir) + if err != nil { + return nil, fmt.Errorf("read block history directory: %w", err) + } + for _, entry := range entries { + if entry.IsDir() || !strings.HasPrefix(entry.Name(), "batch-") || !strings.HasSuffix(entry.Name(), ".json") { + continue + } + through, err := strconv.ParseUint(strings.TrimSuffix(strings.TrimPrefix(entry.Name(), "batch-"), ".json"), 10, 64) + if err != nil { + continue + } + loaded, err := readBatch(filepath.Join(dir, entry.Name())) + if err != nil { + return nil, fmt.Errorf("load block history batch %d: %w", through, err) + } + if loaded.Through != through { + return nil, fmt.Errorf("load block history batch %d: record contains through %d", through, loaded.Through) + } + store.install(loaded) + } + return store, nil +} + +func (s *Store) RecordBlock(block *b.Block) error { + if block == nil { + return errors.New("record block history: nil block") + } + rewards := make([]rpc.BlockReward, len(block.Rewards)) + copy(rewards, block.Rewards) + record := Record{ + Slot: block.Slot, + ParentSlot: block.ParentSlot, + BlockHeight: block.BlockHeight, + Blockhash: solana.Hash(block.Blockhash).String(), + PreviousBlockhash: solana.Hash(block.LastBlockhash).String(), + Rewards: rewards, + } + timestamp := block.UnixTimestamp + if timestamp == 0 && block.FooterProducerTimeNanos != 0 { + timestamp = int64(block.FooterProducerTimeNanos / 1_000_000_000) + } + if timestamp != 0 { + record.BlockTime = ×tamp + } + s.mu.Lock() + s.pending[record.Slot] = record + s.mu.Unlock() + return nil +} + +func (s *Store) RecordSkipped(slot uint64) error { + s.mu.Lock() + s.pending[slot] = Record{Slot: slot, Skipped: true} + s.mu.Unlock() + return nil +} + +// Prepare persists every pending outcome through the fold boundary. It must run +// before the corresponding AccountsDB commit; a leftover unselected batch is +// harmless because Get also checks the recovered rooted watermark. +func (s *Store) Prepare(through uint64) error { + s.mu.RLock() + records := make([]Record, 0) + for slot, record := range s.pending { + if slot <= through { + records = append(records, record) + } + } + s.mu.RUnlock() + if len(records) == 0 { + return nil + } + sort.Slice(records, func(i, j int) bool { return records[i].Slot < records[j].Slot }) + payload := batch{Version: recordVersion, Through: through, Records: records} + tmp, err := os.CreateTemp(s.dir, ".block-history-*") + if err != nil { + return fmt.Errorf("create block history temp file: %w", err) + } + tmpPath := tmp.Name() + defer os.Remove(tmpPath) + writeErr := json.NewEncoder(tmp).Encode(payload) + if writeErr == nil { + writeErr = tmp.Sync() + } + closeErr := tmp.Close() + if writeErr != nil { + return fmt.Errorf("write block history through slot %d: %w", through, writeErr) + } + if closeErr != nil { + return fmt.Errorf("close block history through slot %d: %w", through, closeErr) + } + if err := os.Rename(tmpPath, s.batchPath(through)); err != nil { + return fmt.Errorf("publish block history through slot %d: %w", through, err) + } + if err := syncDir(s.dir); err != nil { + return err + } + s.mu.Lock() + s.install(payload) + s.mu.Unlock() + return nil +} + +func (s *Store) install(value batch) { + slots := make([]uint64, 0, len(value.Records)) + for _, record := range value.Records { + s.persisted[record.Slot] = record + s.sourceBatch[record.Slot] = value.Through + slots = append(slots, record.Slot) + } + s.batches[value.Through] = slots +} + +func (s *Store) SetRooted(slot uint64) error { + return s.setRooted(slot, false) +} + +// Rewind adopts an older durable root and drops every speculative record from +// the abandoned replay attempt before that attempt is run again. +func (s *Store) Rewind(slot uint64) error { + return s.setRooted(slot, true) +} + +// DiscardUnrootedFrom drops a discarded fork suffix without changing the +// durable root. Pending records below from remain available for the next fold. +func (s *Store) DiscardUnrootedFrom(from uint64) error { + s.mu.Lock() + rooted := s.rooted.Load() + if from <= rooted { + s.mu.Unlock() + return fmt.Errorf("cannot discard block history at or below rooted slot %d", rooted) + } + for slot := range s.pending { + if slot >= from { + delete(s.pending, slot) + } + } + removed := false + for through, slots := range s.batches { + if through < from || through <= rooted { + continue + } + if err := os.Remove(s.batchPath(through)); err != nil && !os.IsNotExist(err) { + s.mu.Unlock() + return fmt.Errorf("discard unrooted block history batch %d: %w", through, err) + } + removed = true + for _, slot := range slots { + if s.sourceBatch[slot] == through { + delete(s.persisted, slot) + delete(s.sourceBatch, slot) + } + } + delete(s.batches, through) + } + s.mu.Unlock() + if removed { + return syncDir(s.dir) + } + return nil +} + +func (s *Store) setRooted(slot uint64, rewind bool) error { + s.rooted.Store(slot) + s.mu.Lock() + for pendingSlot := range s.pending { + if rewind || pendingSlot <= slot { + delete(s.pending, pendingSlot) + } + } + pruneBefore := uint64(0) + if slot > s.retentionSlots { + pruneBefore = slot - s.retentionSlots + } + removed := false + for through, slots := range s.batches { + orphaned := through > slot + expired := pruneBefore > 0 && through <= pruneBefore + if !orphaned && !expired { + continue + } + if err := os.Remove(s.batchPath(through)); err != nil && !os.IsNotExist(err) { + s.mu.Unlock() + return fmt.Errorf("prune block history batch %d: %w", through, err) + } + removed = true + for _, recordSlot := range slots { + if s.sourceBatch[recordSlot] == through { + delete(s.persisted, recordSlot) + delete(s.sourceBatch, recordSlot) + } + } + delete(s.batches, through) + } + s.mu.Unlock() + if removed { + if err := syncDir(s.dir); err != nil { + return err + } + } + return nil +} + +func (s *Store) Rooted() uint64 { return s.rooted.Load() } + +func (s *Store) Get(slot uint64) (Record, error) { + if slot > s.rooted.Load() { + return Record{}, ErrNotAvailable + } + s.mu.RLock() + record, ok := s.persisted[slot] + sourceBatch := s.sourceBatch[slot] + s.mu.RUnlock() + if !ok || sourceBatch > s.rooted.Load() { + return Record{}, ErrNotAvailable + } + if record.Skipped { + return Record{}, ErrSlotSkipped + } + return record, nil +} + +func (s *Store) MinimumLedgerSlot() (uint64, error) { return s.first(false) } + +func (s *Store) FirstAvailableBlock() (uint64, error) { return s.first(true) } + +func (s *Store) first(requireBlock bool) (uint64, error) { + rooted := s.rooted.Load() + var first uint64 + found := false + s.mu.RLock() + defer s.mu.RUnlock() + for slot, record := range s.persisted { + if slot > rooted || s.sourceBatch[slot] > rooted || (requireBlock && record.Skipped) { + continue + } + if !found || slot < first { + first = slot + found = true + } + } + if !found { + return 0, ErrNotAvailable + } + return first, nil +} + +func (s *Store) batchPath(through uint64) string { + return filepath.Join(s.dir, fmt.Sprintf("batch-%020d.json", through)) +} + +func readBatch(path string) (batch, error) { + data, err := os.ReadFile(path) + if err != nil { + return batch{}, err + } + var value batch + if err := json.Unmarshal(data, &value); err != nil { + return batch{}, err + } + if value.Version != recordVersion { + return batch{}, fmt.Errorf("unsupported block history version %d", value.Version) + } + return value, nil +} + +func syncDir(path string) error { + dir, err := os.Open(path) + if err != nil { + return fmt.Errorf("open block history directory: %w", err) + } + if err := dir.Sync(); err != nil { + _ = dir.Close() + return fmt.Errorf("sync block history directory: %w", err) + } + if err := dir.Close(); err != nil { + return fmt.Errorf("close block history directory: %w", err) + } + return nil +} diff --git a/pkg/replay/block.go b/pkg/replay/block.go index 495ed53a8..99aac5e49 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -54,6 +54,14 @@ type SlotCtxSetter interface { SetSlotCtx(slotCtx *sealevel.SlotCtx) } +type blockHistoryPublisher interface { + RecordBlockHistory(block *b.Block) error + RecordSkippedBlockHistory(slot uint64) error + PrepareBlockHistory(through uint64) error + SetRootedBlockHistorySlot(slot uint64) error + DiscardUnrootedBlockHistory(from uint64) error +} + // BlockFetchOpts contains options for parallel block fetching type BlockFetchOpts struct { MaxRPS int // Rate limit (requests per second), 0 = use default @@ -1676,6 +1684,7 @@ func ReplayBlocks( onCancelWriteState OnCancelWriteState, // callback to write state immediately on cancellation (can be nil) ) *ReplayResult { result := &ReplayResult{} + historyPublisher, _ := rpcServer.(blockHistoryPublisher) alpenglowMode := consensusOpts != nil && consensusOpts.Alpenglow replayFrontier := uint64(0) if startSlot > 0 { @@ -1993,7 +2002,16 @@ func ReplayBlocks( // only its immutable bytes cross to the async worker. Snapshot: transactionStatuses.SnapshotThrough, Install: func(through uint64, payload []byte) (*state.TransactionStatusCheckpointRef, error) { - return PrepareTransactionStatusCheckpoint(acctsDbPath, through, payload) + ref, err := PrepareTransactionStatusCheckpoint(acctsDbPath, through, payload) + if err != nil { + return nil, err + } + if historyPublisher != nil { + if err := historyPublisher.PrepareBlockHistory(through); err != nil { + return nil, fmt.Errorf("prepare block history: %w", err) + } + } + return ref, nil }, AfterCommit: checkpointAfterCommit, }); hookErr != nil { @@ -2063,6 +2081,11 @@ func ReplayBlocks( mithrilState.LastRootedSlot = promotedThrough mithrilState.LastRootedBankhash = rootedCtx.Bankhash mithrilState.LastRootedContext = rootedCtx + if historyPublisher != nil { + if err := historyPublisher.SetRootedBlockHistorySlot(promotedThrough); err != nil { + mlog.Log.Errorf("block history retention after rooted slot %d: %v", promotedThrough, err) + } + } if transactionStatuses.Root(promotedThrough) { mlog.Log.Infof("transaction status cache reconstructed complete %d-root coverage through durable slot %d", maxTransactionStatusRoots, promotedThrough) @@ -2436,6 +2459,13 @@ func ReplayBlocks( mlog.Log.Warnf("%v — block source rejected the fork rewind", sw) return false } + if historyPublisher != nil { + if err := historyPublisher.DiscardUnrootedBlockHistory(sw.Slot); err != nil { + result.Error = fmt.Errorf("%w: discard unrooted block history: %v", sw, err) + mlog.Log.Warnf("%v", result.Error) + return false + } + } if !parentSwitchNeedsStateUnwind(sw.Slot, currentExecutedAnchorSlot()) { // Trailing skips advance replay's consumed frontier without creating // account-state layers. Finalized ancestry can later require a block @@ -2790,6 +2820,13 @@ func ReplayBlocks( // Handle skipped slots - log and continue without execution if block.IsSkipped { + if historyPublisher != nil { + if err := historyPublisher.RecordSkippedBlockHistory(block.Slot); err != nil { + result.Error = fmt.Errorf("record skipped block history at slot %d: %w", block.Slot, err) + mlog.Log.Errorf("%v", result.Error) + break + } + } // Zero is the explicit locally consumed outcome for a skip. Parent-ID // gap inference is provisional; recording it lets a later certificate // or discovered ancestry require a source rewind and, when necessary, @@ -3024,6 +3061,13 @@ func ReplayBlocks( global.ClearPendingStakePubkeys() break } + if historyPublisher != nil { + if err := historyPublisher.RecordBlockHistory(block); err != nil { + result.Error = fmt.Errorf("record block history at slot %d: %w", block.Slot, err) + mlog.Log.Errorf("%v", result.Error) + break + } + } // The successful child now owns its derived snapshot. Any later bank uses // lastSlotCtx; the one-shot retained unwind bridge is no longer needed. unwoundParentBankSysvars = nil @@ -4295,6 +4339,9 @@ func ProcessBlock( slotCtx.LamportsBurnt = fees.DistributeTxFeesToSlotLeader(acctsDb, slotCtx, block.Leader, &txFeeAccumulator) slotCtx.RecordModifiedAcct(block.Leader) } + if err := recordBlockFeeReward(block, slotCtx, &txFeeAccumulator); err != nil { + return nil, err + } metrics.GlobalBlockReplay.Reward.AddTimingSince(start) start = time.Now() diff --git a/pkg/replay/block_fee_reward.go b/pkg/replay/block_fee_reward.go new file mode 100644 index 000000000..bf11a9f8d --- /dev/null +++ b/pkg/replay/block_fee_reward.go @@ -0,0 +1,52 @@ +package replay + +import ( + "fmt" + "math" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/fees" + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" + "github.com/gagliardetto/solana-go/rpc" +) + +func recordBlockFeeReward(block *b.Block, slotCtx *sealevel.SlotCtx, accumulated *fees.TxFeeInfoAccumulator) error { + if block == nil || slotCtx == nil || accumulated == nil || len(block.Transactions) == 0 || accumulated.TotalFees == 0 { + return nil + } + leader := block.Leader + if !global.ManageLeaderSchedule() && block.BlockReward != nil { + leader = block.BlockReward.Leader + } + if leader == (solana.PublicKey{}) { + return nil + } + if slotCtx.LamportsBurnt > accumulated.TotalFees { + return fmt.Errorf("record fee reward at slot %d: burnt fees %d exceed total fees %d", block.Slot, slotCtx.LamportsBurnt, accumulated.TotalFees) + } + lamports := accumulated.TotalFees - slotCtx.LamportsBurnt + if lamports > math.MaxInt64 { + return fmt.Errorf("record fee reward at slot %d: reward %d overflows int64", block.Slot, lamports) + } + leaderAccount, err := slotCtx.GetAccount(leader) + if err != nil { + return fmt.Errorf("record fee reward at slot %d: read leader %s: %w", block.Slot, leader, err) + } + reward := rpc.BlockReward{ + Pubkey: leader, + Lamports: int64(lamports), + PostBalance: leaderAccount.Lamports, + RewardType: rpc.RewardTypeFee, + } + block.BlockReward = &b.BlockRewardsInfo{Leader: leader, Lamports: lamports, PostBalance: leaderAccount.Lamports} + for i := range block.Rewards { + if block.Rewards[i].RewardType == rpc.RewardTypeFee { + block.Rewards[i] = reward + return nil + } + } + block.Rewards = append(block.Rewards, reward) + return nil +} diff --git a/pkg/replay/leader_finalize.go b/pkg/replay/leader_finalize.go index 78c0b90d9..3a62d0b74 100644 --- a/pkg/replay/leader_finalize.go +++ b/pkg/replay/leader_finalize.go @@ -143,6 +143,9 @@ func CommitLeaderSlot(in CommitLeaderInput) (*sealevel.SlotCtx, error) { slotCtx.LamportsBurnt = fees.DistributeTxFeesToSlotLeader(in.AcctsDb, slotCtx, block.Leader, &in.TxFeeAccumulator) slotCtx.RecordModifiedAcct(block.Leader) } + if err := recordBlockFeeReward(block, slotCtx, &in.TxFeeAccumulator); err != nil { + return nil, err + } var rentSysvar *sealevel.SysvarRent if bankSysvars := slotCtx.BankSysvars(); bankSysvars != nil { if bankRent, ok := bankSysvars.Rent(); ok { diff --git a/pkg/rpcserver/errors.go b/pkg/rpcserver/errors.go index ee2a1654f..2f328d325 100644 --- a/pkg/rpcserver/errors.go +++ b/pkg/rpcserver/errors.go @@ -13,8 +13,12 @@ const ( rpcCodeInvalidParams jsonrpc.ErrorCode = -32602 // -32002 matches Agave's SendTransactionPreflightFailure. rpcCodeSendTransactionPreflightFailure jsonrpc.ErrorCode = -32002 + // -32004 matches Agave's BlockNotAvailable error. + rpcCodeBlockNotAvailable jsonrpc.ErrorCode = -32004 // -32016 is Agave's reserved code for MinContextSlotNotReached. rpcCodeMinContextSlotNotReached jsonrpc.ErrorCode = -32016 + // -32007 matches Agave's SlotSkipped error. + rpcCodeSlotSkipped jsonrpc.ErrorCode = -32007 ) type MinContextSlotNotReachedError struct { @@ -59,6 +63,44 @@ type InvalidParamsError struct { Message string } +type SlotSkippedError struct { + Slot uint64 +} + +type BlockNotAvailableError struct { + Slot uint64 +} + +func (e *BlockNotAvailableError) Error() string { + return fmt.Sprintf("Block not available for slot %d", e.Slot) +} + +func (e *BlockNotAvailableError) ToJSONRPCError() (jsonrpc.JSONRPCError, error) { + return jsonrpc.JSONRPCError{Code: rpcCodeBlockNotAvailable, Message: e.Error()}, nil +} + +func (e *BlockNotAvailableError) FromJSONRPCError(rpcErr jsonrpc.JSONRPCError) error { + if rpcErr.Code != rpcCodeBlockNotAvailable { + return fmt.Errorf("unexpected code %d for BlockNotAvailableError", rpcErr.Code) + } + return nil +} + +func (e *SlotSkippedError) Error() string { + return fmt.Sprintf("Slot %d was skipped", e.Slot) +} + +func (e *SlotSkippedError) ToJSONRPCError() (jsonrpc.JSONRPCError, error) { + return jsonrpc.JSONRPCError{Code: rpcCodeSlotSkipped, Message: e.Error()}, nil +} + +func (e *SlotSkippedError) FromJSONRPCError(rpcErr jsonrpc.JSONRPCError) error { + if rpcErr.Code != rpcCodeSlotSkipped { + return fmt.Errorf("unexpected code %d for SlotSkippedError", rpcErr.Code) + } + return nil +} + func (e *InvalidParamsError) Error() string { return e.Message } func (e *InvalidParamsError) ToJSONRPCError() (jsonrpc.JSONRPCError, error) { @@ -114,6 +156,8 @@ func rpcErrorRegistry() jsonrpc.Errors { errs := jsonrpc.NewErrors() errs.Register(rpcCodeInvalidParams, new(*InvalidParamsError)) errs.Register(rpcCodeSendTransactionPreflightFailure, new(*SendTransactionPreflightFailureError)) + errs.Register(rpcCodeBlockNotAvailable, new(*BlockNotAvailableError)) errs.Register(rpcCodeMinContextSlotNotReached, new(*MinContextSlotNotReachedError)) + errs.Register(rpcCodeSlotSkipped, new(*SlotSkippedError)) return errs } diff --git a/pkg/rpcserver/get_block_history.go b/pkg/rpcserver/get_block_history.go new file mode 100644 index 000000000..a53264fae --- /dev/null +++ b/pkg/rpcserver/get_block_history.go @@ -0,0 +1,150 @@ +package rpcserver + +import ( + "context" + "errors" + "fmt" + "math" + + "github.com/Overclock-Validator/mithril/pkg/blockhistory" + "github.com/filecoin-project/go-jsonrpc" + "github.com/gagliardetto/solana-go/rpc" +) + +type GetBlockResp struct { + BlockHeight uint64 `json:"blockHeight"` + BlockTime *int64 `json:"blockTime"` + Blockhash string `json:"blockhash"` + PreviousBlockhash string `json:"previousBlockhash"` + ParentSlot uint64 `json:"parentSlot"` + Rewards []blockRewardResponse `json:"rewards"` +} + +type blockRewardResponse struct { + rpc.BlockReward + Commission *uint8 `json:"commission"` +} + +func (rpcServer *RpcServer) MinimumLedgerSlot(ctx context.Context, p jsonrpc.RawParams) (uint64, error) { + if err := requireNoParams(p, "minimumLedgerSlot"); err != nil { + return 0, err + } + if rpcServer.blockHistory == nil { + return 0, errors.New("node has no retained block history") + } + return rpcServer.blockHistory.MinimumLedgerSlot() +} + +func (rpcServer *RpcServer) GetFirstAvailableBlock(ctx context.Context, p jsonrpc.RawParams) (uint64, error) { + if err := requireNoParams(p, "getFirstAvailableBlock"); err != nil { + return 0, err + } + if rpcServer.blockHistory == nil { + return 0, errors.New("node has no retained block history") + } + return rpcServer.blockHistory.FirstAvailableBlock() +} + +func (rpcServer *RpcServer) GetBlock(ctx context.Context, p jsonrpc.RawParams) (GetBlockResp, error) { + params, err := jsonrpc.DecodeParams[[]interface{}](p) + if err != nil { + return GetBlockResp{}, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + if len(params) < 1 || len(params) > 2 { + return GetBlockResp{}, &InvalidParamsError{Message: "getBlock requires a slot and optional config"} + } + slot, err := blockRPCUint64(params[0], "slot") + if err != nil { + return GetBlockResp{}, err + } + details, err := validateBlockConfig(params) + if err != nil { + return GetBlockResp{}, err + } + if rpcServer.blockHistory == nil { + return GetBlockResp{}, errors.New("node has no retained block history") + } + if details == "full" { + return GetBlockResp{}, &InvalidParamsError{Message: "getBlock transactionDetails: full is not supported because transactions are not retained"} + } + record, err := rpcServer.blockHistory.Get(slot) + if errors.Is(err, blockhistory.ErrSlotSkipped) { + return GetBlockResp{}, &SlotSkippedError{Slot: slot} + } + if errors.Is(err, blockhistory.ErrNotAvailable) { + return GetBlockResp{}, &BlockNotAvailableError{Slot: slot} + } + if err != nil { + return GetBlockResp{}, err + } + response := GetBlockResp{ + BlockHeight: record.BlockHeight, + BlockTime: record.BlockTime, + Blockhash: record.Blockhash, + PreviousBlockhash: record.PreviousBlockhash, + ParentSlot: record.ParentSlot, + Rewards: make([]blockRewardResponse, len(record.Rewards)), + } + for i, reward := range record.Rewards { + response.Rewards[i] = blockRewardResponse{BlockReward: reward, Commission: reward.Commission} + } + return response, nil +} + +func blockRPCUint64(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 requireNoParams(p jsonrpc.RawParams, method string) error { + if len(p) == 0 { + return nil + } + params, err := jsonrpc.DecodeParams[[]interface{}](p) + if err != nil { + return &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} + } + if len(params) != 0 { + return &InvalidParamsError{Message: method + " does not accept parameters"} + } + return nil +} + +func validateBlockConfig(params []interface{}) (string, error) { + if len(params) == 1 || params[1] == nil { + return "", &InvalidParamsError{Message: "getBlock block history requires transactionDetails: none or full"} + } + config, ok := params[1].(map[string]interface{}) + if !ok { + return "", &InvalidParamsError{Message: "invalid getBlock config"} + } + if raw := config["maxSupportedTransactionVersion"]; raw != nil { + version, err := blockRPCUint64(raw, "maxSupportedTransactionVersion") + if err != nil || version > math.MaxUint8 { + return "", &InvalidParamsError{Message: "invalid maxSupportedTransactionVersion"} + } + } + if commitment, exists := config["commitment"]; exists { + value, ok := commitment.(string) + if !ok || (value != "confirmed" && value != "finalized") { + return "", &InvalidParamsError{Message: "getBlock requires confirmed or finalized commitment"} + } + } + if encoding, exists := config["encoding"]; exists && encoding != "json" { + return "", &InvalidParamsError{Message: "getBlock block history supports json encoding"} + } + details, exists := config["transactionDetails"] + if !exists || (details != "none" && details != "full") { + return "", &InvalidParamsError{Message: "getBlock block history supports transactionDetails: none or full"} + } + if rewards, exists := config["rewards"]; exists { + value, ok := rewards.(bool) + if !ok || !value { + return "", &InvalidParamsError{Message: "getBlock block history requires rewards: true"} + } + } + return details.(string), nil +} diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index 8a3dd7afe..ebc1a9267 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -14,6 +14,8 @@ import ( "time" "github.com/Overclock-Validator/mithril/pkg/accountsdb" + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/blockhistory" "github.com/Overclock-Validator/mithril/pkg/mlog" "github.com/Overclock-Validator/mithril/pkg/sealevel" "github.com/filecoin-project/go-jsonrpc" @@ -31,6 +33,7 @@ type RpcServer struct { slotCtx *sealevel.SlotCtx slotCtxMu sync.RWMutex genesisHash string + blockHistory *blockhistory.Store leaderTPUCacheMu sync.RWMutex leaderTPUByIdentity map[solana.PublicKey]tpuEndpoint @@ -48,14 +51,71 @@ type RpcServer struct { const maxQuietMethodProbeBody = 64 << 10 var supportedRPCMethods = map[string]struct{}{ - "getAccountInfo": {}, - "getBankHash": {}, - "getBlockHeight": {}, - "getEpochInfo": {}, - "getGenesisHash": {}, - "getLatestBlockhash": {}, - "sendTransaction": {}, - "simulateTransaction": {}, + "getAccountInfo": {}, + "getBankHash": {}, + "getBlock": {}, + "getBlockHeight": {}, + "getEpochInfo": {}, + "getFirstAvailableBlock": {}, + "getGenesisHash": {}, + "getLatestBlockhash": {}, + "minimumLedgerSlot": {}, + "sendTransaction": {}, + "simulateTransaction": {}, +} + +func (rpcServer *RpcServer) EnableBlockHistory(dir string, retentionSlots, rootedSlot uint64) error { + store, err := blockhistory.Open(dir, retentionSlots) + if err != nil { + return err + } + if err := store.SetRooted(rootedSlot); err != nil { + return err + } + rpcServer.blockHistory = store + return nil +} + +func (rpcServer *RpcServer) RecordBlockHistory(block *b.Block) error { + if rpcServer == nil || rpcServer.blockHistory == nil { + return nil + } + return rpcServer.blockHistory.RecordBlock(block) +} + +func (rpcServer *RpcServer) RecordSkippedBlockHistory(slot uint64) error { + if rpcServer == nil || rpcServer.blockHistory == nil { + return nil + } + return rpcServer.blockHistory.RecordSkipped(slot) +} + +func (rpcServer *RpcServer) PrepareBlockHistory(through uint64) error { + if rpcServer == nil || rpcServer.blockHistory == nil { + return nil + } + return rpcServer.blockHistory.Prepare(through) +} + +func (rpcServer *RpcServer) SetRootedBlockHistorySlot(slot uint64) error { + if rpcServer == nil || rpcServer.blockHistory == nil { + return nil + } + return rpcServer.blockHistory.SetRooted(slot) +} + +func (rpcServer *RpcServer) RewindBlockHistory(slot uint64) error { + if rpcServer == nil || rpcServer.blockHistory == nil { + return nil + } + return rpcServer.blockHistory.Rewind(slot) +} + +func (rpcServer *RpcServer) DiscardUnrootedBlockHistory(from uint64) error { + if rpcServer == nil || rpcServer.blockHistory == nil { + return nil + } + return rpcServer.blockHistory.DiscardUnrootedFrom(from) } func NewRpcServer(acctsDb *accountsdb.AccountsDb, port uint16, epochSchedule *sealevel.SysvarEpochSchedule, genesisHash solana.Hash) *RpcServer { diff --git a/pkg/snapshot/build_db.go b/pkg/snapshot/build_db.go index a7e2ed934..b1f285ffe 100644 --- a/pkg/snapshot/build_db.go +++ b/pkg/snapshot/build_db.go @@ -51,6 +51,7 @@ func CleanAccountsDbDir(accountsDbDir string) { // so retaining them would waste space and make stale diagnostics look // actionable. "transaction-status-checkpoints", + "rpc-block-history", "mithril_db", "mithril_db_log_shards", "bankhash_db", From 930e27c202e9992f5c826c908e6ccb0dfde50512 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Wed, 23 Sep 2026 23:31:13 +0530 Subject: [PATCH 2/5] test(rpc): cover block retention, fork rewind, and responses --- pkg/blockhistory/store_test.go | 200 ++++++++++++++++++++++++ pkg/replay/block_fee_reward_test.go | 39 +++++ pkg/rpcserver/get_block_history_test.go | 82 ++++++++++ 3 files changed, 321 insertions(+) create mode 100644 pkg/blockhistory/store_test.go create mode 100644 pkg/replay/block_fee_reward_test.go create mode 100644 pkg/rpcserver/get_block_history_test.go diff --git a/pkg/blockhistory/store_test.go b/pkg/blockhistory/store_test.go new file mode 100644 index 000000000..a74616674 --- /dev/null +++ b/pkg/blockhistory/store_test.go @@ -0,0 +1,200 @@ +package blockhistory + +import ( + "errors" + "sync" + "testing" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestStoreConcurrentReadAndPrepare(t *testing.T) { + store, err := Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 1})) + require.NoError(t, store.Prepare(10)) + require.NoError(t, store.SetRooted(10)) + + var readers sync.WaitGroup + readers.Add(1) + go func() { + defer readers.Done() + for i := 0; i < 100000; i++ { + _, _ = store.Get(1) + } + }() + for i := 0; i < 20; i++ { + require.NoError(t, store.RecordBlock(&b.Block{Slot: 1})) + require.NoError(t, store.Prepare(10)) + } + readers.Wait() +} + +func TestStoreOnlyPublishesRootedHistoryAndReplacesFork(t *testing.T) { + store, err := Open(t.TempDir(), 100) + require.NoError(t, err) + + firstHash := solana.Hash{1} + secondHash := solana.Hash{2} + require.NoError(t, store.RecordBlock(&b.Block{Slot: 10, Blockhash: firstHash})) + _, err = store.Get(10) + require.ErrorIs(t, err, ErrNotAvailable) + + require.NoError(t, store.RecordBlock(&b.Block{Slot: 10, Blockhash: secondHash})) + require.NoError(t, store.Prepare(10)) + require.NoError(t, store.SetRooted(10)) + record, err := store.Get(10) + require.NoError(t, err) + require.Equal(t, secondHash.String(), record.Blockhash) +} + +func TestStoreHidesPreparedBatchUntilItsFoldCommits(t *testing.T) { + store, err := Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 11})) + require.NoError(t, store.Prepare(12)) + store.rooted.Store(11) // The account fold through slot 12 has not committed. + _, err = store.Get(11) + require.ErrorIs(t, err, ErrNotAvailable) + _, err = store.FirstAvailableBlock() + require.ErrorIs(t, err, ErrNotAvailable) + require.NoError(t, store.SetRooted(12)) + _, err = store.Get(11) + require.NoError(t, err) +} + +func TestStoreUsesAlpenglowFooterBlockTime(t *testing.T) { + store, err := Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 1, FooterProducerTimeNanos: 1_790_110_767_123_456_789})) + require.NoError(t, store.Prepare(1)) + require.NoError(t, store.SetRooted(1)) + record, err := store.Get(1) + require.NoError(t, err) + require.NotNil(t, record.BlockTime) + require.Equal(t, int64(1_790_110_767), *record.BlockTime) +} + +func TestStorePrepareWithoutPendingHistoryIsNoOp(t *testing.T) { + store, err := Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.Prepare(10)) + require.NoError(t, store.SetRooted(10)) +} + +func TestStoreKeepsLedgerAndBlockFloorsDistinct(t *testing.T) { + store, err := Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.RecordSkipped(8)) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 9})) + require.NoError(t, store.Prepare(9)) + require.NoError(t, store.SetRooted(9)) + + minimum, err := store.MinimumLedgerSlot() + require.NoError(t, err) + require.Equal(t, uint64(8), minimum) + firstBlock, err := store.FirstAvailableBlock() + require.NoError(t, err) + require.Equal(t, uint64(9), firstBlock) + _, err = store.Get(8) + require.ErrorIs(t, err, ErrSlotSkipped) +} + +func TestStoreRetentionPrunesOldRecords(t *testing.T) { + dir := t.TempDir() + store, err := Open(dir, 2) + require.NoError(t, err) + for slot := uint64(1); slot <= 4; slot++ { + require.NoError(t, store.RecordBlock(&b.Block{Slot: slot})) + require.NoError(t, store.Prepare(slot)) + require.NoError(t, store.SetRooted(slot)) + } + _, err = store.Get(1) + require.True(t, errors.Is(err, ErrNotAvailable)) + minimum, err := store.MinimumLedgerSlot() + require.NoError(t, err) + require.Equal(t, uint64(3), minimum) + + reopened, err := Open(dir, 2) + require.NoError(t, err) + require.NoError(t, reopened.SetRooted(4)) + first, err := reopened.FirstAvailableBlock() + require.NoError(t, err) + require.Equal(t, uint64(3), first) +} + +func TestStoreDropsUnselectedBatchAfterRestart(t *testing.T) { + dir := t.TempDir() + store, err := Open(dir, 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 10, Blockhash: solana.Hash{1}})) + require.NoError(t, store.Prepare(10)) // Simulate crash before AccountsDB commits slot 10. + + reopened, err := Open(dir, 100) + require.NoError(t, err) + require.NoError(t, reopened.SetRooted(9)) + require.NoError(t, reopened.RecordBlock(&b.Block{Slot: 10, Blockhash: solana.Hash{2}})) + require.NoError(t, reopened.Prepare(10)) + require.NoError(t, reopened.SetRooted(10)) + record, err := reopened.Get(10) + require.NoError(t, err) + require.Equal(t, solana.Hash{2}.String(), record.Blockhash) +} + +func TestStoreRewindDropsPersistedAndSpeculativeForkHistory(t *testing.T) { + dir := t.TempDir() + store, err := Open(dir, 100) + require.NoError(t, err) + for slot := uint64(10); slot <= 12; slot++ { + require.NoError(t, store.RecordBlock(&b.Block{Slot: slot, Blockhash: solana.Hash{byte(slot)}})) + require.NoError(t, store.Prepare(slot)) + require.NoError(t, store.SetRooted(slot)) + } + require.NoError(t, store.RecordBlock(&b.Block{Slot: 13, Blockhash: solana.Hash{13}})) + + require.NoError(t, store.Rewind(10)) + _, err = store.Get(11) + require.ErrorIs(t, err, ErrNotAvailable) + + // A replacement fold must not inherit the abandoned slot 13 record. + require.NoError(t, store.RecordBlock(&b.Block{Slot: 11, Blockhash: solana.Hash{21}})) + require.NoError(t, store.Prepare(13)) + require.NoError(t, store.SetRooted(13)) + _, err = store.Get(13) + require.ErrorIs(t, err, ErrNotAvailable) + record, err := store.Get(11) + require.NoError(t, err) + require.Equal(t, solana.Hash{21}.String(), record.Blockhash) +} + +func TestStoreDiscardsUnrootedForkSuffixWithoutAdvancingRoot(t *testing.T) { + dir := t.TempDir() + store, err := Open(dir, 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 10, Blockhash: solana.Hash{10}})) + require.NoError(t, store.Prepare(10)) + require.NoError(t, store.SetRooted(10)) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 11, Blockhash: solana.Hash{11}})) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 12, Blockhash: solana.Hash{12}})) + require.NoError(t, store.Prepare(12)) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 13, Blockhash: solana.Hash{13}})) + + require.NoError(t, store.DiscardUnrootedFrom(12)) + require.Equal(t, uint64(10), store.Rooted()) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 12, Blockhash: solana.Hash{22}})) + require.NoError(t, store.Prepare(13)) + require.NoError(t, store.SetRooted(13)) + + reopened, err := Open(dir, 100) + require.NoError(t, err) + require.NoError(t, reopened.SetRooted(13)) + for slot, hash := range map[uint64]solana.Hash{10: {10}, 11: {11}, 12: {22}} { + record, err := reopened.Get(slot) + require.NoError(t, err) + require.Equal(t, hash.String(), record.Blockhash) + } + _, err = reopened.Get(13) + require.ErrorIs(t, err, ErrNotAvailable) +} diff --git a/pkg/replay/block_fee_reward_test.go b/pkg/replay/block_fee_reward_test.go new file mode 100644 index 000000000..59b4eefa7 --- /dev/null +++ b/pkg/replay/block_fee_reward_test.go @@ -0,0 +1,39 @@ +package replay + +import ( + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/fees" + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" + "github.com/gagliardetto/solana-go/rpc" + "github.com/stretchr/testify/require" +) + +func TestRecordBlockFeeRewardUsesExecutedBankResult(t *testing.T) { + global.SetManageLeaderSchedule(true) + defer global.SetManageLeaderSchedule(false) + leader := solana.PublicKey{1} + mem := accounts.NewMemAccounts() + require.NoError(t, mem.SetAccountWithoutLock(leader, &accounts.Account{Key: leader, Lamports: 1_025})) + slotCtx := &sealevel.SlotCtx{Slot: 7, Accounts: mem, LamportsBurnt: 25} + block := &b.Block{Slot: 7, Leader: leader, Transactions: []*solana.Transaction{{}}} + + require.NoError(t, recordBlockFeeReward(block, slotCtx, &fees.TxFeeInfoAccumulator{TotalFees: 50})) + require.Len(t, block.Rewards, 1) + require.Equal(t, rpc.RewardTypeFee, block.Rewards[0].RewardType) + require.Equal(t, int64(25), block.Rewards[0].Lamports) + require.Equal(t, uint64(1_025), block.Rewards[0].PostBalance) + require.Equal(t, uint64(25), block.BlockReward.Lamports) +} + +func TestRecordBlockFeeRewardIgnoresBlockWithoutLeader(t *testing.T) { + require.NoError(t, recordBlockFeeReward( + &b.Block{Slot: 8, Transactions: []*solana.Transaction{{}}}, + &sealevel.SlotCtx{Slot: 8}, + &fees.TxFeeInfoAccumulator{TotalFees: 50}, + )) +} diff --git a/pkg/rpcserver/get_block_history_test.go b/pkg/rpcserver/get_block_history_test.go new file mode 100644 index 000000000..ba245363b --- /dev/null +++ b/pkg/rpcserver/get_block_history_test.go @@ -0,0 +1,82 @@ +package rpcserver + +import ( + "encoding/json" + "testing" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/blockhistory" + "github.com/filecoin-project/go-jsonrpc" + "github.com/gagliardetto/solana-go" + "github.com/gagliardetto/solana-go/rpc" + "github.com/stretchr/testify/require" +) + +func TestBlockHistoryRPCServesExporterContract(t *testing.T) { + store, err := blockhistory.Open(t.TempDir(), 100) + require.NoError(t, err) + leader := solana.PublicKey{1} + require.NoError(t, store.RecordSkipped(40)) + require.NoError(t, store.RecordBlock(&b.Block{ + Slot: 41, + ParentSlot: 39, + BlockHeight: 30, + Blockhash: solana.Hash{2}, + LastBlockhash: solana.Hash{3}, + Rewards: []rpc.BlockReward{{ + Pubkey: leader, Lamports: 25, PostBalance: 1_025, RewardType: rpc.RewardTypeFee, + }}, + })) + require.NoError(t, store.Prepare(41)) + require.NoError(t, store.SetRooted(41)) + server := &RpcServer{blockHistory: store} + + minimum, err := server.MinimumLedgerSlot(t.Context(), jsonrpc.RawParams(`[]`)) + require.NoError(t, err) + require.Equal(t, uint64(40), minimum) + first, err := server.GetFirstAvailableBlock(t.Context(), jsonrpc.RawParams(`[]`)) + require.NoError(t, err) + require.Equal(t, uint64(41), first) + for _, params := range []jsonrpc.RawParams{nil, jsonrpc.RawParams(`null`)} { + minimum, err = server.MinimumLedgerSlot(t.Context(), params) + require.NoError(t, err) + require.Equal(t, uint64(40), minimum) + first, err = server.GetFirstAvailableBlock(t.Context(), params) + require.NoError(t, err) + require.Equal(t, uint64(41), first) + } + + response, err := server.GetBlock(t.Context(), jsonrpc.RawParams(`[41,{"commitment":"confirmed","encoding":"json","transactionDetails":"none","rewards":true,"maxSupportedTransactionVersion":0}]`)) + require.NoError(t, err) + require.Equal(t, uint64(30), response.BlockHeight) + require.Equal(t, int64(25), response.Rewards[0].Lamports) + require.Equal(t, leader, response.Rewards[0].Pubkey) + encoded, err := json.Marshal(response) + require.NoError(t, err) + require.Contains(t, string(encoded), `"commission":null`) + + _, err = server.GetBlock(t.Context(), jsonrpc.RawParams(`[40,{"transactionDetails":"none"}]`)) + var skipped *SlotSkippedError + require.ErrorAs(t, err, &skipped) + require.Equal(t, uint64(40), skipped.Slot) + + _, err = server.GetBlock(t.Context(), jsonrpc.RawParams(`[42,{"transactionDetails":"none"}]`)) + var unavailable *BlockNotAvailableError + require.ErrorAs(t, err, &unavailable) + require.Equal(t, uint64(42), unavailable.Slot) +} + +func TestGetBlockRejectsUnretainedTransactionDetails(t *testing.T) { + store, err := blockhistory.Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 1})) + require.NoError(t, store.Prepare(1)) + require.NoError(t, store.SetRooted(1)) + server := &RpcServer{blockHistory: store} + _, err = server.GetBlock(t.Context(), jsonrpc.RawParams(`[1,{"transactionDetails":"full"}]`)) + var invalid *InvalidParamsError + require.ErrorAs(t, err, &invalid) + require.Contains(t, invalid.Error(), "transactions are not retained") + _, err = server.GetBlock(t.Context(), jsonrpc.RawParams(`[2,{"transactionDetails":"full"}]`)) + require.ErrorAs(t, err, &invalid) +} From 56469d05bd121bc360508d0b9ea703677b44995c Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Wed, 30 Sep 2026 23:40:53 +0530 Subject: [PATCH 3/5] fix(rpc): keep block history consistent across commits and rewinds --- .../node/block_history_recovery_test.go | 147 ++++++++++++++++++ cmd/mithril/node/node.go | 19 ++- cmd/mithril/node/rpc_block_history.go | 10 ++ cmd/mithril/node/rpc_block_history_test.go | 43 +++++ pkg/blockhistory/recovery_test.go | 31 ++++ pkg/blockhistory/store.go | 47 ++++++ pkg/replay/block.go | 31 +++- pkg/replay/block_history.go | 20 +++ pkg/replay/block_history_flush_test.go | 71 +++++++++ pkg/replay/block_history_test.go | 104 +++++++++++++ pkg/rpcserver/get_block_history.go | 15 ++ pkg/rpcserver/history_recovery_test.go | 92 +++++++++++ pkg/rpcserver/rpc_registration_test.go | 91 +++++++++++ pkg/rpcserver/rpcserver.go | 36 ++++- 14 files changed, 747 insertions(+), 10 deletions(-) create mode 100644 cmd/mithril/node/block_history_recovery_test.go create mode 100644 cmd/mithril/node/rpc_block_history.go create mode 100644 cmd/mithril/node/rpc_block_history_test.go create mode 100644 pkg/blockhistory/recovery_test.go create mode 100644 pkg/replay/block_history.go create mode 100644 pkg/replay/block_history_flush_test.go create mode 100644 pkg/replay/block_history_test.go create mode 100644 pkg/rpcserver/history_recovery_test.go create mode 100644 pkg/rpcserver/rpc_registration_test.go diff --git a/cmd/mithril/node/block_history_recovery_test.go b/cmd/mithril/node/block_history_recovery_test.go new file mode 100644 index 000000000..9bcb8f051 --- /dev/null +++ b/cmd/mithril/node/block_history_recovery_test.go @@ -0,0 +1,147 @@ +package node + +import ( + "bytes" + "encoding/json" + "io" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + "github.com/Overclock-Validator/mithril/pkg/accountsdb" + "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/replay" + "github.com/Overclock-Validator/mithril/pkg/rpcserver" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/Overclock-Validator/mithril/pkg/state" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestBlockHistoryHTTPAfterDurableRewind(t *testing.T) { + for _, name := range []string{"rewind", "adoption failure", "prune failure", "accounts cleanup failure", "accounts precommit failure"} { + t.Run(name, func(t *testing.T) { + 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) + address := solana.PublicKey{91} + var tip *state.ResumeContext + var checkpoint string + for _, slot := range []uint64{2, 4} { + tip = testDurableStatusContext(t, dir, slot) + if slot == 2 { + checkpoint = filepath.Join(dir, replay.TransactionStatusCheckpointDirectory, tip.TransactionStatusCheckpoint.File) + } + encoded, err := json.Marshal(tip) + require.NoError(t, err) + _, err = db.CommitBatch([]accounts.SlotDelta{{Slot: slot, Delta: []*accounts.Account{{Key: address, Lamports: slot}}}}, slot, nil, encoded) + require.NoError(t, err) + } + server := rpcserver.NewRpcServer(db, 0, &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 3}, solana.Hash{1}) + require.NoError(t, server.EnableBlockHistory(filepath.Join(dir, "rpc-block-history"), 100, 0)) + require.NoError(t, server.RecordBlockHistory(&block.Block{Slot: 4, ParentSlot: 2, Blockhash: solana.Hash{4}, LastBlockhash: solana.Hash{2}})) + require.NoError(t, server.PrepareBlockHistory(4)) + require.NoError(t, server.SetRootedBlockHistorySlot(4)) + endpoint := httptest.NewServer(server) + t.Cleanup(endpoint.Close) + endpoint.Client().Timeout = time.Second + blockRequest := `{"jsonrpc":"2.0","id":1,"method":"getBlock","params":[4,{"transactionDetails":"none"}]}` + call := func(request string) map[string]json.RawMessage { + t.Helper() + response, err := endpoint.Client().Post(endpoint.URL, "application/json", bytes.NewBufferString(request)) + require.NoError(t, err) + defer response.Body.Close() + payload, err := io.ReadAll(response.Body) + require.NoError(t, err) + t.Logf("HTTP=%d response=%s", response.StatusCode, payload) + var result map[string]json.RawMessage + require.NoError(t, json.Unmarshal(payload, &result)) + return result + } + require.Contains(t, string(call(blockRequest)["result"]), solana.Hash{4}.String()) + s := &state.MithrilState{LastRootedSlot: 4, LastRootedContext: tip, LastSlot: 4} + var publisher replay.SlotCtxSetter = server + if name == "adoption failure" { + publisher = &historyAdoptionFault{RpcServer: server, checkpoint: checkpoint} + } + if name == "prune failure" { + batchPath := filepath.Join(dir, "rpc-block-history", "batch-00000000000000000004.json") + require.NoError(t, os.Remove(batchPath)) + require.NoError(t, os.Mkdir(batchPath, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(batchPath, "blocker"), nil, 0o644)) + } + var accountFaultPath string + if name == "accounts cleanup failure" { + accountFaultPath = filepath.Join(db.AcctsDir, "rewound") + require.NoError(t, os.WriteFile(accountFaultPath, nil, 0o644)) + } + if name == "accounts precommit failure" { + manifests, err := accountsdb.ListFoldManifests(db.AcctsDir) + require.NoError(t, err) + for _, manifest := range manifests { + if manifest.ThroughSlot == 4 { + accountFaultPath = manifest.Path + ".rewound" + } + } + require.NotEmpty(t, accountFaultPath) + require.NoError(t, os.Mkdir(accountFaultPath, 0o755)) + } + require.Equal(t, name == "rewind", rewindStoreBelowDivergence(dir, db, s, 3, 100, publisher)) + if name == "adoption failure" { + require.NoFileExists(t, checkpoint) + require.Equal(t, uint64(4), s.LastRootedSlot, "adoption failed after the account root changed") + } + account, err := db.GetAccount(2, address) + require.NoError(t, err) + if name == "accounts precommit failure" { + require.Equal(t, uint64(4), account.Lamports, "failed manifest parking must not alter accounts") + } else { + require.Equal(t, uint64(2), account.Lamports) + } + if name != "rewind" { + for _, request := range []string{ + blockRequest, + `{"jsonrpc":"2.0","id":1,"method":"minimumLedgerSlot","params":[]}`, + `{"jsonrpc":"2.0","id":1,"method":"getFirstAvailableBlock","params":[]}`, + } { + reply := call(request) + require.Empty(t, reply["result"]) + require.Contains(t, string(reply["error"]), "RPC history is unavailable during account recovery") + } + if accountFaultPath == "" { + // Inspect the side-store watermark independently of the gate; recovery remains halted in production. + server.SetHistoryRecoveryPending(false) + require.JSONEq(t, `{"code":-32004,"message":"Block not available for slot 4"}`, string(call(blockRequest)["error"])) + return + } + require.FileExists(t, filepath.Join(dir, "rpc-block-history", "batch-00000000000000000004.json"), "failed account rewind must retain history") + require.NoError(t, os.Remove(accountFaultPath)) + require.True(t, rewindStoreBelowDivergence(dir, db, s, 3, 100, server)) + } + require.Equal(t, uint64(2), s.LastRootedSlot) + require.JSONEq(t, `{"code":-32004,"message":"Block not available for slot 4"}`, string(call(blockRequest)["error"])) + }) + } +} + +// Remove the checkpoint after the real DB rewind and history update, before adoption validates it again. +type historyAdoptionFault struct { + *rpcserver.RpcServer + checkpoint string +} + +func (f *historyAdoptionFault) RewindBlockHistory(slot uint64) error { + if err := f.RpcServer.RewindBlockHistory(slot); err != nil { + return err + } + return os.Remove(f.checkpoint) +} diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index 2bf22f378..103c781a0 100644 --- a/cmd/mithril/node/node.go +++ b/cmd/mithril/node/node.go @@ -2567,7 +2567,7 @@ postBootstrap: } else if rpcPort != 0 { rpcServer = rpcserver.NewRpcServer(accountsDb, uint16(rpcPort), epochScheduleFromState(mithrilState), solana.MustHashFromBase58(networkGenesisHash)) retentionSlots := epochScheduleFromState(mithrilState).SlotsPerEpoch * 2 - if err := rpcServer.EnableBlockHistory(filepath.Join(accountsPath, "rpc-block-history"), retentionSlots, mithrilState.LastRootedSlot); err != nil { + if err := enableRPCBlockHistory(rpcServer, alpenglowMode, filepath.Join(accountsPath, "rpc-block-history"), retentionSlots, mithrilState.LastRootedSlot); err != nil { klog.Fatalf("enable RPC block history: %v", err) } rpcServer.Start() @@ -4337,7 +4337,11 @@ func completeInterruptedRewind(accountsDbPath string, accountsDb *accountsdb.Acc // boundary strictly below divSlot and adopts that boundary's context as the // rooted checkpoint. Returns false (caller halts) when no boundary below the // divergence is retained or the rewind/reconcile fails. -func rewindStoreBelowDivergence(accountsDbPath string, accountsDb *accountsdb.AccountsDb, s *state.MithrilState, divSlot uint64, rewindHorizon uint64) bool { +func rewindStoreBelowDivergence(accountsDbPath string, accountsDb *accountsdb.AccountsDb, s *state.MithrilState, divSlot uint64, rewindHorizon uint64, rpcServer replay.SlotCtxSetter) bool { + historyGate, hasHistoryGate := rpcServer.(interface{ SetHistoryRecoveryPending(bool) }) + if hasHistoryGate { + historyGate.SetHistoryRecoveryPending(true) + } points, err := accountsDb.ListRewindPoints() if err != nil { mlog.Log.Errorf("fork switch: cannot list rewind boundaries: %v; halting", err) @@ -4368,12 +4372,21 @@ func rewindStoreBelowDivergence(accountsDbPath string, accountsDb *accountsdb.Ac mlog.Log.Errorf("fork switch: rewind to boundary %d failed: %v; halting", target, err) return false } + if history, ok := rpcServer.(interface{ RewindBlockHistory(uint64) error }); ok { + if err := history.RewindBlockHistory(res.NewThrough); err != nil { + mlog.Log.Errorf("fork switch: rewind RPC block history to slot %d: %v; halting", res.NewThrough, err) + return false + } + } if err := adoptRewindResult(accountsDbPath, s, res); err != nil { mlog.Log.Errorf("fork switch: %v; halting", err) return false } mlog.Log.Warnf("fork switch: superseded slot %d was already folded (R=%d); rewound durable state to boundary %d (%d batches / %d keys undone)", divSlot, oldRooted, res.NewThrough, res.UndoneBatches, res.UndoneKeys) + if hasHistoryGate { + historyGate.SetHistoryRecoveryPending(false) + } return true } @@ -4602,7 +4615,7 @@ func runReplayWithRecovery( // (this is exactly why durable state deliberately lags and undo pointers // are retained for a horizon). Beyond the horizon -> fail closed. if divSlot <= mithrilState.LastRootedSlot { - if !rewindStoreBelowDivergence(accountsDbPath, accountsDb, mithrilState, divSlot, rewindHorizon) { + if !rewindStoreBelowDivergence(accountsDbPath, accountsDb, mithrilState, divSlot, rewindHorizon, rpcServer) { break } } diff --git a/cmd/mithril/node/rpc_block_history.go b/cmd/mithril/node/rpc_block_history.go new file mode 100644 index 000000000..1f8e33951 --- /dev/null +++ b/cmd/mithril/node/rpc_block_history.go @@ -0,0 +1,10 @@ +package node + +import "github.com/Overclock-Validator/mithril/pkg/rpcserver" + +func enableRPCBlockHistory(server *rpcserver.RpcServer, alpenglowMode bool, dir string, retentionSlots, rootedSlot uint64) error { + if !alpenglowMode { + return nil + } + return server.EnableBlockHistory(dir, retentionSlots, rootedSlot) +} diff --git a/cmd/mithril/node/rpc_block_history_test.go b/cmd/mithril/node/rpc_block_history_test.go new file mode 100644 index 000000000..418b2cc79 --- /dev/null +++ b/cmd/mithril/node/rpc_block_history_test.go @@ -0,0 +1,43 @@ +package node + +import ( + "context" + "os" + "path/filepath" + "testing" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/rpcserver" + "github.com/stretchr/testify/require" +) + +func TestRPCBlockHistoryOnlyStartsOnAlpenglow(t *testing.T) { + for _, alpenglowMode := range []bool{false, true} { + name := "classic" + if alpenglowMode { + name = "alpenglow" + } + t.Run(name, func(t *testing.T) { + server := &rpcserver.RpcServer{} + dir := filepath.Join(t.TempDir(), "rpc-block-history") + require.NoError(t, enableRPCBlockHistory(server, alpenglowMode, dir, 10, 0)) + for slot := uint64(1); slot <= 1000; slot++ { + require.NoError(t, server.RecordBlockHistory(&b.Block{Slot: slot})) + if alpenglowMode && slot%10 == 0 { + require.NoError(t, server.PrepareBlockHistory(slot)) + require.NoError(t, server.SetRootedBlockHistorySlot(slot)) + } + } + if !alpenglowMode { + _, err := server.MinimumLedgerSlot(context.Background(), nil) + require.ErrorContains(t, err, "no retained block history") + _, err = os.Stat(dir) + require.ErrorIs(t, err, os.ErrNotExist) + return + } + first, err := server.GetFirstAvailableBlock(context.Background(), nil) + require.NoError(t, err) + require.Equal(t, uint64(991), first) + }) + } +} diff --git a/pkg/blockhistory/recovery_test.go b/pkg/blockhistory/recovery_test.go new file mode 100644 index 000000000..865723859 --- /dev/null +++ b/pkg/blockhistory/recovery_test.go @@ -0,0 +1,31 @@ +package blockhistory + +import ( + "os" + "testing" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestStoreRestoresCommittedRecordAfterOrphanReadRetry(t *testing.T) { + store, err := Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 5, Blockhash: solana.Hash{5}})) + require.NoError(t, store.Prepare(6)) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 5, Blockhash: solana.Hash{55}})) + require.NoError(t, store.Prepare(8)) + + committed := store.batchPath(6) + require.NoError(t, os.Rename(committed, committed+".saved")) + require.Error(t, store.SetRooted(6)) + require.FileExists(t, store.batchPath(8), "a failed recovery must leave the orphan available for retry") + _, err = store.Get(5) + require.ErrorIs(t, err, ErrNotAvailable) + require.NoError(t, os.Rename(committed+".saved", committed)) + require.NoError(t, store.SetRooted(6)) + record, err := store.Get(5) + require.NoError(t, err) + require.Equal(t, solana.Hash{5}.String(), record.Blockhash) +} diff --git a/pkg/blockhistory/store.go b/pkg/blockhistory/store.go index 2d34fb56f..93e4201ca 100644 --- a/pkg/blockhistory/store.go +++ b/pkg/blockhistory/store.go @@ -209,6 +209,15 @@ func (s *Store) DiscardUnrootedFrom(from uint64) error { } } removed := false + for through := range s.batches { + if through >= from { + if err := s.restoreRetainedRecords(0, from-1); err != nil { + s.mu.Unlock() + return err + } + break + } + } for through, slots := range s.batches { if through < from || through <= rooted { continue @@ -246,6 +255,15 @@ func (s *Store) setRooted(slot uint64, rewind bool) error { pruneBefore = slot - s.retentionSlots } removed := false + for through := range s.batches { + if through > slot { + if err := s.restoreRetainedRecords(pruneBefore, slot); err != nil { + s.mu.Unlock() + return err + } + break + } + } for through, slots := range s.batches { orphaned := through > slot expired := pruneBefore > 0 && through <= pruneBefore @@ -274,6 +292,35 @@ func (s *Store) setRooted(slot uint64, rewind bool) error { return nil } +// An interrupted multi-chunk fold can leave the same slot in several batches. +// Removing an uncommitted batch must reveal the preceding committed record. +func (s *Store) restoreRetainedRecords(pruneBefore, throughLimit uint64) error { + throughs := make([]uint64, 0, len(s.batches)) + for through := range s.batches { + if through <= throughLimit && (pruneBefore == 0 || through > pruneBefore) { + throughs = append(throughs, through) + } + } + sort.Slice(throughs, func(i, j int) bool { return throughs[i] < throughs[j] }) + persisted := make(map[uint64]Record) + sources := make(map[uint64]uint64) + for _, through := range throughs { + value, err := readBatch(s.batchPath(through)) + if err != nil { + return fmt.Errorf("restore block history batch %d: %w", through, err) + } + if value.Through != through { + return fmt.Errorf("restore block history batch %d: record contains through %d", through, value.Through) + } + for _, record := range value.Records { + persisted[record.Slot] = record + sources[record.Slot] = through + } + } + s.persisted, s.sourceBatch = persisted, sources + return nil +} + func (s *Store) Rooted() uint64 { return s.rooted.Load() } func (s *Store) Get(slot uint64) (Record, error) { diff --git a/pkg/replay/block.go b/pkg/replay/block.go index 1aec50e02..0cc5130ea 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -2026,7 +2026,17 @@ func ReplayBlocks( } return ref, nil }, - AfterCommit: checkpointAfterCommit, + AfterCommit: func(ref *state.TransactionStatusCheckpointRef) error { + var err error + if historyPublisher != nil { + // Forced flushes may commit several chunks before replay applies their results. + err = historyPublisher.SetRootedBlockHistorySlot(ref.Root) + } + if checkpointAfterCommit != nil { + err = errors.Join(err, checkpointAfterCommit(ref)) + } + return err + }, }); hookErr != nil { result.Error = fmt.Errorf("configure durable transaction status checkpoints: %w", hookErr) return result @@ -2994,8 +3004,15 @@ func ReplayBlocks( currentSlot = block.Slot block.Epoch = epochSchedule.GetEpoch(currentSlot) - var configErr error initialBlockConfigured := lastSlotCtx == nil + var historyParentBlockhash solana.Hash + if historyPublisher != nil && initialBlockConfigured { + historyParentBlockhash = initialLastBlockhash + if resumeState != nil { + historyParentBlockhash = solana.Hash(resumeState.LastBlockhash) + } + } + var configErr error // Use lastSlotCtx == nil to detect first block, not currentSlot == startSlot. // This handles the case where startSlot (or slots after it) are skipped - // the first emitted block might have slot > startSlot. @@ -3193,10 +3210,12 @@ func ReplayBlocks( break } if historyPublisher != nil { - if err := historyPublisher.RecordBlockHistory(block); err != nil { - result.Error = fmt.Errorf("record block history at slot %d: %w", block.Slot, err) - mlog.Log.Errorf("%v", result.Error) - break + if historyBlock := blockForHistory(block, historyParentBlockhash); historyBlock != nil { + if err := historyPublisher.RecordBlockHistory(historyBlock); err != nil { + result.Error = fmt.Errorf("record block history at slot %d: %w", block.Slot, err) + mlog.Log.Errorf("%v", result.Error) + break + } } } // The successful child now owns its derived snapshot. Any later bank uses diff --git a/pkg/replay/block_history.go b/pkg/replay/block_history.go new file mode 100644 index 000000000..68ae6c2e1 --- /dev/null +++ b/pkg/replay/block_history.go @@ -0,0 +1,20 @@ +package replay + +import ( + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/gagliardetto/solana-go" +) + +// blockForHistory supplies metadata from the executed parent without changing +// the block's execution context. Unknown non-genesis parents remain unavailable. +func blockForHistory(block *b.Block, parentBlockhash solana.Hash) *b.Block { + if block.LastBlockhash != ([32]byte{}) || block.Slot == 0 { + return block + } + if parentBlockhash == (solana.Hash{}) { + return nil + } + historyBlock := *block + historyBlock.LastBlockhash = parentBlockhash + return &historyBlock +} diff --git a/pkg/replay/block_history_flush_test.go b/pkg/replay/block_history_flush_test.go new file mode 100644 index 000000000..952d8120f --- /dev/null +++ b/pkg/replay/block_history_flush_test.go @@ -0,0 +1,71 @@ +package replay + +import ( + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accounts" + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/blockhistory" + "github.com/Overclock-Validator/mithril/pkg/state" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestBlockHistorySurvivesPartialForcedFlush(t *testing.T) { + for _, publishEachCommit := range []bool{false, true} { + for _, restart := range []bool{false, true} { + name := "running" + if publishEachCommit { + name += "_publish_each_commit" + } + if restart { + name += "_restart" + } + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + store, err := blockhistory.Open(dir, 100) + require.NoError(t, err) + require.NoError(t, store.SetRooted(4)) + for _, slot := range []uint64{5, 6, 7, 8} { + require.NoError(t, store.RecordBlock(&b.Block{Slot: slot, ParentSlot: slot - 1, Blockhash: solana.Hash{byte(slot)}, LastBlockhash: solana.Hash{byte(slot - 1)}})) + } + committer := &fakeCommitter{durable: accounts.NewMemAccounts(), failOn: 7} + tail := asyncTestTail(committer, 5, 6, 7, 8) + statuses := NewTransactionStatusCache() + checkpointDir := t.TempDir() + var prepared []uint64 + require.NoError(t, tail.SetTransactionStatusCheckpointHooks(TransactionStatusCheckpointHooks{ + Capture: statuses.CaptureSnapshotThrough, + AfterCommit: func(ref *state.TransactionStatusCheckpointRef) error { + if publishEachCommit { + return store.SetRooted(ref.Root) + } + return nil + }, + Install: func(through uint64, payload []byte) (*state.TransactionStatusCheckpointRef, error) { + ref, err := PrepareTransactionStatusCheckpoint(checkpointDir, through, payload) + if err != nil { + return nil, err + } + prepared = append(prepared, through) + return ref, store.Prepare(through) + }, + })) + through, rooted, err := tail.flush(8) + require.ErrorContains(t, err, "commit boom at slot 7") + require.Equal(t, []uint64{6, 8}, prepared) + require.Equal(t, []uint64{6}, committer.throughs) + require.Equal(t, uint64(6), through) + require.Equal(t, uint64(6), rooted.Slot) + if restart { + store, err = blockhistory.Open(dir, 100) + require.NoError(t, err) + } + require.NoError(t, store.SetRooted(through)) + got, err := store.Get(5) + require.NoError(t, err, "first fold committed slot 5 before second fold failed") + require.Equal(t, solana.Hash{5}.String(), got.Blockhash) + }) + } + } +} diff --git a/pkg/replay/block_history_test.go b/pkg/replay/block_history_test.go new file mode 100644 index 000000000..4fa9435f3 --- /dev/null +++ b/pkg/replay/block_history_test.go @@ -0,0 +1,104 @@ +package replay + +import ( + "math" + "testing" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/blockhistory" + "github.com/Overclock-Validator/mithril/pkg/epochstakes" + "github.com/Overclock-Validator/mithril/pkg/global" + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/Overclock-Validator/mithril/pkg/state" + "github.com/Overclock-Validator/mithril/pkg/turbine" + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestFreshBlockHistoryPreservesSnapshotParentHash(t *testing.T) { + const epoch, parentSlot = uint64(100_000), uint64(800_000) + vote := solana.PublicKey{249} + db := epochBoundaryStakeDB(t, parentSlot, []sealevel.Delegation{{ + VoterPubkey: vote, StakeLamports: 1, + ActivationEpoch: math.MaxUint64, DeactivationEpoch: math.MaxUint64, + }}) + require.False(t, global.HasEpochStakes(epoch)) + global.PutEpochStakes(epoch, map[solana.PublicKey]uint64{vote: 1}, map[solana.PublicKey]*epochstakes.VoteAccount{vote: {NodePubkey: vote}}, 1) + t.Cleanup(func() { global.ClearEpochStakes(epoch); global.SetLeaderScheduleForEpoch(epoch, nil) }) + previousManage := global.ManageLeaderSchedule() + global.SetManageLeaderSchedule(true) + t.Cleanup(func() { global.SetManageLeaderSchedule(previousManage) }) + parentHash := solana.Hash{77} + firstHash := solana.Hash{88} + manifestState := &state.MithrilState{ + ManifestParentSlot: parentSlot, + ManifestParentBankhash: (solana.Hash{66}).String(), + ManifestEvictedBlockhash: (solana.Hash{55}).String(), + ManifestFeeRateGovernor: &state.ManifestFeeRateGovernorSeed{}, + ManifestRecentBlockhashes: []state.BlockhashEntry{{Blockhash: parentHash.String()}}, + } + schedule := &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 8, LeaderScheduleSlotOffset: 8} + // The first executed block can follow skipped slots; its parent remains the snapshot. + first := turbine.BlockFromEntries(parentSlot+4, parentSlot, []turbine.Entry{{Hash: firstHash}}) + first.Epoch = epoch + require.NoError(t, configureInitialBlock(db, first, manifestState, schedule)) + second := &b.Block{Slot: parentSlot + 5, Epoch: epoch, Blockhash: solana.Hash{99}} + require.NoError(t, configureBlock(second, &sealevel.SlotCtx{Slot: first.Slot, Epoch: epoch, Blockhash: first.Blockhash}, schedule)) + require.Equal(t, [32]byte(firstHash), second.LastBlockhash) + dir := t.TempDir() + store, err := blockhistory.Open(dir, 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(blockForHistory(first, parentHash))) + require.NoError(t, store.RecordBlock(blockForHistory(second, firstHash))) + require.Zero(t, first.LastBlockhash, "history must not alter the execution block's nonce context") + require.NoError(t, store.Prepare(second.Slot)) + require.NoError(t, store.SetRooted(second.Slot)) + reopened, err := blockhistory.Open(dir, 100) + require.NoError(t, err) + require.NoError(t, reopened.SetRooted(second.Slot)) + firstAvailable, err := reopened.FirstAvailableBlock() + require.NoError(t, err) + require.Equal(t, first.Slot, firstAvailable) + secondRecord, err := reopened.Get(second.Slot) + require.NoError(t, err) + require.Equal(t, firstHash.String(), secondRecord.PreviousBlockhash) + firstRecord, err := reopened.Get(firstAvailable) + require.NoError(t, err) + require.Equal(t, parentHash.String(), firstRecord.PreviousBlockhash, + "fresh snapshot parent hash must survive first replay block recording, persistence and restart") +} + +func TestBlockHistoryUnknownParentRemainsUnavailable(t *testing.T) { + const parentSlot = uint64(100) + first := &b.Block{Slot: parentSlot + 1, ParentSlot: parentSlot, Blockhash: solana.Hash{1}} + second := &b.Block{Slot: parentSlot + 2, ParentSlot: first.Slot, Blockhash: solana.Hash{2}, LastBlockhash: first.Blockhash} + store, err := blockhistory.Open(t.TempDir(), 100) + require.NoError(t, err) + require.Nil(t, blockForHistory(first, solana.Hash{})) + require.NoError(t, store.RecordBlock(blockForHistory(second, solana.Hash{}))) + require.NoError(t, store.Prepare(second.Slot)) + require.NoError(t, store.SetRooted(second.Slot)) + _, err = store.Get(first.Slot) + require.ErrorIs(t, err, blockhistory.ErrNotAvailable) + firstAvailable, err := store.FirstAvailableBlock() + require.NoError(t, err) + require.Equal(t, second.Slot, firstAvailable) + record, err := store.Get(second.Slot) + require.NoError(t, err) + require.Equal(t, solana.Hash(first.Blockhash).String(), record.PreviousBlockhash) + require.Zero(t, first.LastBlockhash) + require.Equal(t, first.Blockhash, second.LastBlockhash) + + knownParent := solana.Hash{3} + second.LastBlockhash = knownParent + require.Same(t, second, blockForHistory(second, first.Blockhash), "an existing parent hash must not be replaced") + require.Equal(t, [32]byte(knownParent), second.LastBlockhash) + genesis := &b.Block{} + require.Same(t, genesis, blockForHistory(genesis, solana.Hash{})) + childOfGenesis := &b.Block{Slot: 1, ParentSlot: 0} + require.Nil(t, blockForHistory(childOfGenesis, solana.Hash{}), "a genesis child still requires a known parent hash") + historyChild := blockForHistory(childOfGenesis, knownParent) + require.NotNil(t, historyChild) + require.Equal(t, [32]byte(knownParent), historyChild.LastBlockhash) + require.Zero(t, childOfGenesis.LastBlockhash) +} diff --git a/pkg/rpcserver/get_block_history.go b/pkg/rpcserver/get_block_history.go index a53264fae..dce9c7c21 100644 --- a/pkg/rpcserver/get_block_history.go +++ b/pkg/rpcserver/get_block_history.go @@ -26,6 +26,11 @@ type blockRewardResponse struct { } func (rpcServer *RpcServer) MinimumLedgerSlot(ctx context.Context, p jsonrpc.RawParams) (uint64, error) { + rpcServer.historyRecoveryMu.RLock() + defer rpcServer.historyRecoveryMu.RUnlock() + if rpcServer.historyRecoveryPending { + return 0, errors.New("RPC history is unavailable during account recovery") + } if err := requireNoParams(p, "minimumLedgerSlot"); err != nil { return 0, err } @@ -36,6 +41,11 @@ func (rpcServer *RpcServer) MinimumLedgerSlot(ctx context.Context, p jsonrpc.Raw } func (rpcServer *RpcServer) GetFirstAvailableBlock(ctx context.Context, p jsonrpc.RawParams) (uint64, error) { + rpcServer.historyRecoveryMu.RLock() + defer rpcServer.historyRecoveryMu.RUnlock() + if rpcServer.historyRecoveryPending { + return 0, errors.New("RPC history is unavailable during account recovery") + } if err := requireNoParams(p, "getFirstAvailableBlock"); err != nil { return 0, err } @@ -46,6 +56,11 @@ func (rpcServer *RpcServer) GetFirstAvailableBlock(ctx context.Context, p jsonrp } func (rpcServer *RpcServer) GetBlock(ctx context.Context, p jsonrpc.RawParams) (GetBlockResp, error) { + rpcServer.historyRecoveryMu.RLock() + defer rpcServer.historyRecoveryMu.RUnlock() + if rpcServer.historyRecoveryPending { + return GetBlockResp{}, errors.New("RPC history is unavailable during account recovery") + } params, err := jsonrpc.DecodeParams[[]interface{}](p) if err != nil { return GetBlockResp{}, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} diff --git a/pkg/rpcserver/history_recovery_test.go b/pkg/rpcserver/history_recovery_test.go new file mode 100644 index 000000000..322f9022e --- /dev/null +++ b/pkg/rpcserver/history_recovery_test.go @@ -0,0 +1,92 @@ +package rpcserver + +import ( + "reflect" + "runtime" + "testing" + "time" + + b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/filecoin-project/go-jsonrpc" + "github.com/stretchr/testify/require" +) + +func TestHistoryRecoveryWaitsForReaders(t *testing.T) { + var disabled *RpcServer + disabled.SetHistoryRecoveryPending(true) + disabled.SetHistoryRecoveryPending(false) + + _, registered := reflect.TypeOf(rpcHandler{}).MethodByName("SetHistoryRecoveryPending") + require.False(t, registered, "the internal recovery gate must not be an RPC method") + + server := &RpcServer{} + server.historyRecoveryMu.RLock() + readerHeld := true + defer func() { + if readerHeld { + server.historyRecoveryMu.RUnlock() + } + }() + done := make(chan struct{}) + go func() { + server.SetHistoryRecoveryPending(true) + close(done) + }() + + // Observe the waiting writer without releasing the in-flight reader. + timeout := time.After(5 * time.Second) + for server.historyRecoveryMu.TryRLock() { + server.historyRecoveryMu.RUnlock() + select { + case <-timeout: + t.Fatal("recovery did not wait for the in-flight reader") + default: + runtime.Gosched() + } + } + select { + case <-done: + t.Fatal("recovery closed the gate before the in-flight reader finished") + default: + } + server.historyRecoveryMu.RUnlock() + readerHeld = false + select { + case <-done: + case <-timeout: + t.Fatal("recovery remained blocked after the reader finished") + } + require.True(t, server.historyRecoveryPending) + server.SetHistoryRecoveryPending(false) + require.False(t, server.historyRecoveryPending) +} + +func TestBlockHistoryRecoveryGate(t *testing.T) { + server := &RpcServer{} + require.NoError(t, server.EnableBlockHistory(t.TempDir(), 100, 0)) + require.NoError(t, server.RecordBlockHistory(&b.Block{Slot: 2, ParentSlot: 1, BlockHeight: 2})) + require.NoError(t, server.PrepareBlockHistory(2)) + require.NoError(t, server.SetRootedBlockHistorySlot(2)) + + for _, pending := range []bool{false, true, false} { + server.SetHistoryRecoveryPending(pending) + minimum, minimumErr := server.MinimumLedgerSlot(t.Context(), jsonrpc.RawParams("[]")) + first, firstErr := server.GetFirstAvailableBlock(t.Context(), jsonrpc.RawParams("[]")) + block, blockErr := server.GetBlock(t.Context(), jsonrpc.RawParams(`[2,{"transactionDetails":"none"}]`)) + if pending { + for _, err := range []error{minimumErr, firstErr, blockErr} { + require.EqualError(t, err, "RPC history is unavailable during account recovery") + } + require.Zero(t, minimum) + require.Zero(t, first) + require.Zero(t, block) + continue + } + require.NoError(t, minimumErr) + require.NoError(t, firstErr) + require.NoError(t, blockErr) + require.Equal(t, uint64(2), minimum) + require.Equal(t, uint64(2), first) + require.Equal(t, uint64(2), block.BlockHeight) + } +} diff --git a/pkg/rpcserver/rpc_registration_test.go b/pkg/rpcserver/rpc_registration_test.go new file mode 100644 index 000000000..cf1756a3a --- /dev/null +++ b/pkg/rpcserver/rpc_registration_test.go @@ -0,0 +1,91 @@ +package rpcserver + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/sealevel" + "github.com/gagliardetto/solana-go" +) + +func TestRPCHandlerMethodSet(t *testing.T) { + typ := reflect.TypeOf(rpcHandler{}) + if typ.NumMethod() != len(supportedRPCMethods) { + t.Fatalf("registered %d methods, public method list has %d", typ.NumMethod(), len(supportedRPCMethods)) + } + for i := 0; i < typ.NumMethod(); i++ { + name := typ.Method(i).Name + name = strings.ToLower(name[:1]) + name[1:] + if _, ok := supportedRPCMethods[name]; !ok { + t.Fatalf("non-public method registered: %s", name) + } + } +} + +func TestInternalMethodsAreNotRPC(t *testing.T) { + for _, mode := range []string{"single", "mixed-batch", "oversized-single", "direct-dispatch"} { + t.Run(mode, func(t *testing.T) { + rpc := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 32}, solana.Hash{1}) + t.Cleanup(func() { _ = rpc.listener.Close() }) + directory := filepath.Join(t.TempDir(), "history") + call := map[string]any{"jsonrpc": "2.0", "id": 1, "method": "enableBlockHistory", "params": []any{directory, 100, 0}} + if mode == "oversized-single" { + call["padding"] = strings.Repeat("x", maxQuietMethodProbeBody+1) + } + var request any = call + if mode == "mixed-batch" { + request = []any{map[string]any{"jsonrpc": "2.0", "id": 2, "method": "getGenesisHash", "params": []any{}}, call} + } + body, err := json.Marshal(request) + if err != nil { + t.Fatal(err) + } + response := httptest.NewRecorder() + if mode == "direct-dispatch" { + rpc.rpcService.HandleRequest(context.Background(), bytes.NewReader(body), response) + } else { + rpc.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))) + } + var replies []struct { + ID int `json:"id"` + Result string `json:"result"` + Error *struct { + Code int `json:"code"` + } `json:"error"` + } + result := response.Body.Bytes() + if mode != "mixed-batch" { + result = append(append([]byte{'['}, result...), ']') + } + if err := json.Unmarshal(result, &replies); err != nil { + t.Fatalf("decode response %q: %v", result, err) + } + found := false + for _, reply := range replies { + if reply.ID == 1 { + found = true + if reply.Error == nil || reply.Error.Code != -32601 { + t.Fatalf("internal method was not rejected: %s", result) + } + } + if reply.ID == 2 && (reply.Error != nil || reply.Result != (solana.Hash{1}).String()) { + t.Fatalf("public batch method failed: %s", result) + } + } + if !found || (mode == "mixed-batch" && len(replies) != 2) { + t.Fatalf("missing response: %s", result) + } + if _, err := os.Stat(directory); !os.IsNotExist(err) { + t.Fatalf("internal request created history directory: %v", err) + } + }) + } +} diff --git a/pkg/rpcserver/rpcserver.go b/pkg/rpcserver/rpcserver.go index ebc1a9267..78f73ac1e 100644 --- a/pkg/rpcserver/rpcserver.go +++ b/pkg/rpcserver/rpcserver.go @@ -23,7 +23,30 @@ import ( "github.com/gagliardetto/solana-go" ) +// rpcHandler exposes only public RPC methods to go-jsonrpc's reflection-based +// registration. Internal setters and lifecycle hooks remain Go-only. +type rpcHandler struct { + rpcMethods +} + +type rpcMethods interface { + GetAccountInfo(context.Context, jsonrpc.RawParams) (GetAccountInfoResp, error) + GetBankHash(context.Context, jsonrpc.RawParams) (string, error) + GetBlock(context.Context, jsonrpc.RawParams) (GetBlockResp, error) + GetBlockHeight(context.Context, jsonrpc.RawParams) (uint64, error) + GetEpochInfo(context.Context, jsonrpc.RawParams) (GetEpochInfoResp, error) + GetFirstAvailableBlock(context.Context, jsonrpc.RawParams) (uint64, error) + GetGenesisHash(context.Context, jsonrpc.RawParams) (string, error) + GetLatestBlockhash(context.Context, jsonrpc.RawParams) (GetLatestBlockhashResp, error) + MinimumLedgerSlot(context.Context, jsonrpc.RawParams) (uint64, error) + SendTransaction(context.Context, jsonrpc.RawParams) (string, error) + SimulateTransaction(context.Context, jsonrpc.RawParams) (SimulateTransactionResp, error) +} + type RpcServer struct { + historyRecoveryMu sync.RWMutex + historyRecoveryPending bool + isReady bool rpcService *jsonrpc.RPCServer serv *httptest.Server @@ -118,6 +141,17 @@ func (rpcServer *RpcServer) DiscardUnrootedBlockHistory(from uint64) error { return rpcServer.blockHistory.DiscardUnrootedFrom(from) } +// SetHistoryRecoveryPending blocks history reads during a durable account rewind. +// Failed recovery leaves the gate closed without deleting retained records. +func (rpcServer *RpcServer) SetHistoryRecoveryPending(pending bool) { + if rpcServer == nil { + return + } + rpcServer.historyRecoveryMu.Lock() + defer rpcServer.historyRecoveryMu.Unlock() + rpcServer.historyRecoveryPending = pending +} + func NewRpcServer(acctsDb *accountsdb.AccountsDb, port uint16, epochSchedule *sealevel.SysvarEpochSchedule, genesisHash solana.Hash) *RpcServer { var err error rpcServer := &RpcServer{genesisHash: genesisHash.String()} @@ -137,7 +171,7 @@ func NewRpcServer(acctsDb *accountsdb.AccountsDb, port uint16, epochSchedule *se jsonrpc.WithServerErrors(rpcErrors), ) - rpcServer.rpcService.Register("MithrilRpc", rpcServer) + rpcServer.rpcService.Register("MithrilRpc", rpcHandler{rpcServer}) rpcServer.acctsDb = acctsDb if epochSchedule != nil { rpcServer.epochSchedule = epochSchedule From cd5ba5e5010f197ab29b797201a3a6f3d76c9990 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Tue, 13 Oct 2026 02:09:15 +0530 Subject: [PATCH 4/5] Record block fee rewards for the executed collector --- pkg/replay/block_fee_reward.go | 28 +++++++------ pkg/replay/block_fee_reward_test.go | 62 ++++++++++++++++++++++++++++- 2 files changed, 75 insertions(+), 15 deletions(-) diff --git a/pkg/replay/block_fee_reward.go b/pkg/replay/block_fee_reward.go index bf11a9f8d..9edcb47fc 100644 --- a/pkg/replay/block_fee_reward.go +++ b/pkg/replay/block_fee_reward.go @@ -6,7 +6,6 @@ import ( b "github.com/Overclock-Validator/mithril/pkg/block" "github.com/Overclock-Validator/mithril/pkg/fees" - "github.com/Overclock-Validator/mithril/pkg/global" "github.com/Overclock-Validator/mithril/pkg/sealevel" "github.com/gagliardetto/solana-go" "github.com/gagliardetto/solana-go/rpc" @@ -16,31 +15,34 @@ func recordBlockFeeReward(block *b.Block, slotCtx *sealevel.SlotCtx, accumulated if block == nil || slotCtx == nil || accumulated == nil || len(block.Transactions) == 0 || accumulated.TotalFees == 0 { return nil } - leader := block.Leader - if !global.ManageLeaderSchedule() && block.BlockReward != nil { - leader = block.BlockReward.Leader - } - if leader == (solana.PublicKey{}) { - return nil - } if slotCtx.LamportsBurnt > accumulated.TotalFees { return fmt.Errorf("record fee reward at slot %d: burnt fees %d exceed total fees %d", block.Slot, slotCtx.LamportsBurnt, accumulated.TotalFees) } lamports := accumulated.TotalFees - slotCtx.LamportsBurnt + if lamports == 0 { + return nil + } if lamports > math.MaxInt64 { return fmt.Errorf("record fee reward at slot %d: reward %d overflows int64", block.Slot, lamports) } - leaderAccount, err := slotCtx.GetAccount(leader) + collector, _, err := transactionFeeCollector(block) + if err != nil { + return fmt.Errorf("record fee reward at slot %d: resolve collector: %w", block.Slot, err) + } + if collector == (solana.PublicKey{}) { + return nil + } + collectorAccount, err := slotCtx.GetAccount(collector) if err != nil { - return fmt.Errorf("record fee reward at slot %d: read leader %s: %w", block.Slot, leader, err) + return fmt.Errorf("record fee reward at slot %d: read collector %s: %w", block.Slot, collector, err) } reward := rpc.BlockReward{ - Pubkey: leader, + Pubkey: collector, Lamports: int64(lamports), - PostBalance: leaderAccount.Lamports, + PostBalance: collectorAccount.Lamports, RewardType: rpc.RewardTypeFee, } - block.BlockReward = &b.BlockRewardsInfo{Leader: leader, Lamports: lamports, PostBalance: leaderAccount.Lamports} + block.BlockReward = &b.BlockRewardsInfo{Leader: collector, Lamports: lamports, PostBalance: collectorAccount.Lamports} for i := range block.Rewards { if block.Rewards[i].RewardType == rpc.RewardTypeFee { block.Rewards[i] = reward diff --git a/pkg/replay/block_fee_reward_test.go b/pkg/replay/block_fee_reward_test.go index 59b4eefa7..b79040849 100644 --- a/pkg/replay/block_fee_reward_test.go +++ b/pkg/replay/block_fee_reward_test.go @@ -1,10 +1,13 @@ package replay import ( + "sync" "testing" "github.com/Overclock-Validator/mithril/pkg/accounts" b "github.com/Overclock-Validator/mithril/pkg/block" + "github.com/Overclock-Validator/mithril/pkg/epochstakes" + "github.com/Overclock-Validator/mithril/pkg/features" "github.com/Overclock-Validator/mithril/pkg/fees" "github.com/Overclock-Validator/mithril/pkg/global" "github.com/Overclock-Validator/mithril/pkg/sealevel" @@ -20,7 +23,7 @@ func TestRecordBlockFeeRewardUsesExecutedBankResult(t *testing.T) { mem := accounts.NewMemAccounts() require.NoError(t, mem.SetAccountWithoutLock(leader, &accounts.Account{Key: leader, Lamports: 1_025})) slotCtx := &sealevel.SlotCtx{Slot: 7, Accounts: mem, LamportsBurnt: 25} - block := &b.Block{Slot: 7, Leader: leader, Transactions: []*solana.Transaction{{}}} + block := &b.Block{Slot: 7, Leader: leader, Features: features.NewFeaturesDefault(), Transactions: []*solana.Transaction{{}}} require.NoError(t, recordBlockFeeReward(block, slotCtx, &fees.TxFeeInfoAccumulator{TotalFees: 50})) require.Len(t, block.Rewards, 1) @@ -32,8 +35,63 @@ func TestRecordBlockFeeRewardUsesExecutedBankResult(t *testing.T) { func TestRecordBlockFeeRewardIgnoresBlockWithoutLeader(t *testing.T) { require.NoError(t, recordBlockFeeReward( - &b.Block{Slot: 8, Transactions: []*solana.Transaction{{}}}, + &b.Block{Slot: 8, Features: features.NewFeaturesDefault(), Transactions: []*solana.Transaction{{}}}, &sealevel.SlotCtx{Slot: 8}, &fees.TxFeeInfoAccumulator{TotalFees: 50}, )) } + +func TestRecordBlockFeeRewardUsesCustomCollector(t *testing.T) { + for _, tc := range []struct { + name string + identityLoaded bool + burn bool + }{ + {"identity loaded", true, false}, + {"identity not loaded", false, false}, + {"invalid collector burns fees", false, true}, + } { + t.Run(tc.name, func(t *testing.T) { + leader, vote, collector := solana.PublicKey{201}, solana.PublicKey{202}, solana.PublicKey{203} + global.SetLeaderSchedule(nil) + global.PutEpochStakes(987, map[solana.PublicKey]uint64{vote: 1}, map[solana.PublicKey]*epochstakes.VoteAccount{vote: {NodePubkey: leader, BlockRevenueCollector: &collector}}, 1) + t.Cleanup(func() { global.ClearEpochStakes(987); global.SetLeaderSchedule(nil) }) + f := features.NewFeaturesDefault() + f.EnableFeature(features.Alpenglow, 0) + f.EnableFeature(features.CustomCommissionCollector, 0) + f.EnableFeature(features.RewardFullPriorityFee, 0) + mem := accounts.NewMemAccounts() + account := &accounts.Account{Key: collector, Lamports: 10_000_000} + if tc.burn { + account.Owner = solana.PublicKey{204} + } + require.NoError(t, mem.SetAccountWithoutLock(collector, account)) + if tc.identityLoaded { + require.NoError(t, mem.SetAccountWithoutLock(leader, &accounts.Account{Key: leader, Lamports: 1_000_000})) + } + ctx := &sealevel.SlotCtx{Slot: 9870, Epoch: 987, Features: f, Accounts: mem, ParentAccts: accounts.NewMemAccounts(), AcctMapsMu: new(sync.Mutex), ModifiedAccts: map[solana.PublicKey]bool{}, WritableAccts: map[solana.PublicKey]bool{}} + rent := sealevel.NewDefaultRentSysvar() + sysvars, err := sealevel.NewBankSysvars(ctx.Slot, &accounts.Account{Key: sealevel.SysvarRentAddr, Lamports: 1, Data: rent.MustMarshal()}) + require.NoError(t, err) + require.NoError(t, ctx.PublishBankSysvars(sysvars)) + block := &b.Block{Slot: ctx.Slot, Epoch: ctx.Epoch, Leader: leader, Features: f, Transactions: []*solana.Transaction{{}}} + accumulated := &fees.TxFeeInfoAccumulator{TotalFees: 50, PriorityFees: 50} + require.NoError(t, distributeBlockTxFees(ctx, block, accumulated)) + paid, err := ctx.GetAccount(collector) + require.NoError(t, err) + wantBalance := uint64(10_000_050) + if tc.burn { + wantBalance = 10_000_000 + require.Equal(t, uint64(50), ctx.LamportsBurnt) + } + require.Equal(t, wantBalance, paid.Lamports) + require.NoError(t, recordBlockFeeReward(block, ctx, accumulated)) + if tc.burn { + require.Empty(t, block.Rewards) + return + } + require.Equal(t, []rpc.BlockReward{{Pubkey: collector, Lamports: 50, PostBalance: paid.Lamports, RewardType: rpc.RewardTypeFee}}, block.Rewards) + require.Equal(t, collector, block.BlockReward.Leader) + }) + } +} From 331b3fd3a4ee410c32f6981c816adcbeac8ee11f Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Sun, 18 Oct 2026 23:11:35 +0530 Subject: [PATCH 5/5] fix(rpc): honor getBlock options and preserve slot precision --- pkg/rpcserver/get_block_history.go | 113 +++++++++++---------- pkg/rpcserver/get_block_history_test.go | 127 ++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 50 deletions(-) diff --git a/pkg/rpcserver/get_block_history.go b/pkg/rpcserver/get_block_history.go index dce9c7c21..6c5b442aa 100644 --- a/pkg/rpcserver/get_block_history.go +++ b/pkg/rpcserver/get_block_history.go @@ -2,9 +2,9 @@ package rpcserver import ( "context" + "encoding/json" "errors" "fmt" - "math" "github.com/Overclock-Validator/mithril/pkg/blockhistory" "github.com/filecoin-project/go-jsonrpc" @@ -17,7 +17,7 @@ type GetBlockResp struct { Blockhash string `json:"blockhash"` PreviousBlockhash string `json:"previousBlockhash"` ParentSlot uint64 `json:"parentSlot"` - Rewards []blockRewardResponse `json:"rewards"` + Rewards []blockRewardResponse `json:"rewards,omitzero"` } type blockRewardResponse struct { @@ -61,33 +61,33 @@ func (rpcServer *RpcServer) GetBlock(ctx context.Context, p jsonrpc.RawParams) ( if rpcServer.historyRecoveryPending { return GetBlockResp{}, errors.New("RPC history is unavailable during account recovery") } - params, err := jsonrpc.DecodeParams[[]interface{}](p) + params, err := jsonrpc.DecodeParams[[]json.RawMessage](p) if err != nil { return GetBlockResp{}, &InvalidParamsError{Message: fmt.Sprintf("decoding params: %v", err)} } if len(params) < 1 || len(params) > 2 { return GetBlockResp{}, &InvalidParamsError{Message: "getBlock requires a slot and optional config"} } - slot, err := blockRPCUint64(params[0], "slot") - if err != nil { - return GetBlockResp{}, err + var slot *uint64 + if err := json.Unmarshal(params[0], &slot); err != nil || slot == nil { + return GetBlockResp{}, &InvalidParamsError{Message: "invalid slot"} } - details, err := validateBlockConfig(params) + details, showRewards, err := validateBlockConfig(params) if err != nil { return GetBlockResp{}, err } if rpcServer.blockHistory == nil { return GetBlockResp{}, errors.New("node has no retained block history") } - if details == "full" { - return GetBlockResp{}, &InvalidParamsError{Message: "getBlock transactionDetails: full is not supported because transactions are not retained"} + if details != "none" { + return GetBlockResp{}, &InvalidParamsError{Message: fmt.Sprintf("getBlock transactionDetails: %s is not supported because transactions are not retained", details)} } - record, err := rpcServer.blockHistory.Get(slot) + record, err := rpcServer.blockHistory.Get(*slot) if errors.Is(err, blockhistory.ErrSlotSkipped) { - return GetBlockResp{}, &SlotSkippedError{Slot: slot} + return GetBlockResp{}, &SlotSkippedError{Slot: *slot} } if errors.Is(err, blockhistory.ErrNotAvailable) { - return GetBlockResp{}, &BlockNotAvailableError{Slot: slot} + return GetBlockResp{}, &BlockNotAvailableError{Slot: *slot} } if err != nil { return GetBlockResp{}, err @@ -98,22 +98,16 @@ func (rpcServer *RpcServer) GetBlock(ctx context.Context, p jsonrpc.RawParams) ( Blockhash: record.Blockhash, PreviousBlockhash: record.PreviousBlockhash, ParentSlot: record.ParentSlot, - Rewards: make([]blockRewardResponse, len(record.Rewards)), } - for i, reward := range record.Rewards { - response.Rewards[i] = blockRewardResponse{BlockReward: reward, Commission: reward.Commission} + if showRewards { + response.Rewards = make([]blockRewardResponse, len(record.Rewards)) + for i, reward := range record.Rewards { + response.Rewards[i] = blockRewardResponse{BlockReward: reward, Commission: reward.Commission} + } } return response, nil } -func blockRPCUint64(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 requireNoParams(p jsonrpc.RawParams, method string) error { if len(p) == 0 { return nil @@ -128,38 +122,57 @@ func requireNoParams(p jsonrpc.RawParams, method string) error { return nil } -func validateBlockConfig(params []interface{}) (string, error) { - if len(params) == 1 || params[1] == nil { - return "", &InvalidParamsError{Message: "getBlock block history requires transactionDetails: none or full"} - } - config, ok := params[1].(map[string]interface{}) - if !ok { - return "", &InvalidParamsError{Message: "invalid getBlock config"} - } - if raw := config["maxSupportedTransactionVersion"]; raw != nil { - version, err := blockRPCUint64(raw, "maxSupportedTransactionVersion") - if err != nil || version > math.MaxUint8 { - return "", &InvalidParamsError{Message: "invalid maxSupportedTransactionVersion"} +func validateBlockConfig(params []json.RawMessage) (string, bool, error) { + var config struct { + Encoding *string `json:"encoding"` + TransactionDetails *string `json:"transactionDetails"` + Rewards *bool `json:"rewards"` + Commitment *string `json:"commitment"` + MaxSupportedTransactionVersion *uint8 `json:"maxSupportedTransactionVersion"` + } + if len(params) == 2 { + if err := json.Unmarshal(params[1], &config); err != nil { + var fieldError *json.UnmarshalTypeError + if errors.As(err, &fieldError) && fieldError.Field != "" { + return "", false, &InvalidParamsError{Message: "invalid " + fieldError.Field} + } + // The legacy config form is an encoding string. + var encoding string + if err := json.Unmarshal(params[1], &encoding); err != nil { + return "", false, &InvalidParamsError{Message: "invalid getBlock config"} + } + config.Encoding = &encoding } } - if commitment, exists := config["commitment"]; exists { - value, ok := commitment.(string) - if !ok || (value != "confirmed" && value != "finalized") { - return "", &InvalidParamsError{Message: "getBlock requires confirmed or finalized commitment"} + if config.Commitment != nil { + if value := *config.Commitment; value != "confirmed" && value != "finalized" { + return "", false, &InvalidParamsError{Message: "getBlock requires confirmed or finalized commitment"} } } - if encoding, exists := config["encoding"]; exists && encoding != "json" { - return "", &InvalidParamsError{Message: "getBlock block history supports json encoding"} + encoding := "json" + if config.Encoding != nil { + encoding = *config.Encoding } - details, exists := config["transactionDetails"] - if !exists || (details != "none" && details != "full") { - return "", &InvalidParamsError{Message: "getBlock block history supports transactionDetails: none or full"} + switch encoding { + case "binary", "base58", "base64", "json", "jsonParsed": + default: + return "", false, &InvalidParamsError{Message: "invalid encoding"} } - if rewards, exists := config["rewards"]; exists { - value, ok := rewards.(bool) - if !ok || !value { - return "", &InvalidParamsError{Message: "getBlock block history requires rewards: true"} - } + if (encoding == "binary" || encoding == "base58") && config.MaxSupportedTransactionVersion != nil && *config.MaxSupportedTransactionVersion >= 1 { + return "", false, &InvalidParamsError{Message: "base58 encoding is not supported with maxSupportedTransactionVersion >= 1"} + } + details := "full" + if config.TransactionDetails != nil { + details = *config.TransactionDetails + } + switch details { + case "full", "none", "signatures", "accounts": + default: + return "", false, &InvalidParamsError{Message: "invalid transactionDetails"} + } + showRewards := true + if config.Rewards != nil { + showRewards = *config.Rewards } - return details.(string), nil + return details, showRewards, nil } diff --git a/pkg/rpcserver/get_block_history_test.go b/pkg/rpcserver/get_block_history_test.go index ba245363b..cfddd0f86 100644 --- a/pkg/rpcserver/get_block_history_test.go +++ b/pkg/rpcserver/get_block_history_test.go @@ -2,10 +2,14 @@ package rpcserver import ( "encoding/json" + "net/http" + "net/http/httptest" + "strings" "testing" b "github.com/Overclock-Validator/mithril/pkg/block" "github.com/Overclock-Validator/mithril/pkg/blockhistory" + "github.com/Overclock-Validator/mithril/pkg/sealevel" "github.com/filecoin-project/go-jsonrpc" "github.com/gagliardetto/solana-go" "github.com/gagliardetto/solana-go/rpc" @@ -80,3 +84,126 @@ func TestGetBlockRejectsUnretainedTransactionDetails(t *testing.T) { _, err = server.GetBlock(t.Context(), jsonrpc.RawParams(`[2,{"transactionDetails":"full"}]`)) require.ErrorAs(t, err, &invalid) } + +func TestGetBlockRewardsHTTP(t *testing.T) { + store, err := blockhistory.Open(t.TempDir(), 100) + require.NoError(t, err) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 1, BlockHeight: 30, Rewards: []rpc.BlockReward{{ + Pubkey: solana.PublicKey{}, Lamports: 25, PostBalance: 1_025, RewardType: rpc.RewardTypeFee, + }}})) + require.NoError(t, store.RecordBlock(&b.Block{Slot: 2, BlockHeight: 30})) + require.NoError(t, store.Prepare(2)) + require.NoError(t, store.SetRooted(2)) + server := NewRpcServer(nil, 0, &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 32}, solana.Hash{}) + require.NoError(t, server.listener.Close()) + server.blockHistory = store + httpServer := httptest.NewServer(server) + t.Cleanup(httpServer.Close) + + const rewards = `[{"pubkey":"11111111111111111111111111111111","lamports":25,"postBalance":1025,"rewardType":"Fee","commission":null}]` + const fullUnavailable = "getBlock transactionDetails: full is not supported because transactions are not retained" + // Agave 1407e9c6 defaults to full/JSON/rewards=true. None does not encode + // transactions, but encoding enums and the base58/version constraint still + // validate. This summary-only server must reject, not downgrade, full. + for _, test := range []struct { + name string + params string + wantRewards string + wantError string + wantInvalid bool + wantCode int + }{ + {name: "default", params: `[1,{"transactionDetails":"none"}]`, wantRewards: rewards}, + {name: "true", params: `[1,{"transactionDetails":"none","rewards":true}]`, wantRewards: rewards}, + {name: "false", params: `[1,{"transactionDetails":"none","rewards":false}]`}, + {name: "empty-default", params: `[2,{"transactionDetails":"none"}]`, wantRewards: `[]`}, + {name: "empty-true", params: `[2,{"transactionDetails":"none","rewards":true}]`, wantRewards: `[]`}, + {name: "empty-false", params: `[2,{"transactionDetails":"none","rewards":false}]`}, + {name: "invalid-rewards", params: `[1,{"transactionDetails":"none","rewards":"false"}]`, wantError: "invalid rewards"}, + {name: "full-without-rewards", params: `[1,{"transactionDetails":"full","rewards":false}]`, wantError: fullUnavailable}, + {name: "exporter-none-json", params: `[1,{"commitment":"confirmed","encoding":"json","transactionDetails":"none","rewards":true,"maxSupportedTransactionVersion":0}]`, wantRewards: rewards}, + {name: "none-finalized", params: `[1,{"commitment":"finalized","transactionDetails":"none"}]`, wantRewards: rewards}, + {name: "none-base64", params: `[1,{"transactionDetails":"none","encoding":"base64"}]`, wantRewards: rewards}, + {name: "none-base58", params: `[1,{"transactionDetails":"none","encoding":"base58","maxSupportedTransactionVersion":0}]`, wantRewards: rewards}, + {name: "none-binary", params: `[1,{"transactionDetails":"none","encoding":"binary"}]`, wantRewards: rewards}, + {name: "none-json-parsed", params: `[1,{"transactionDetails":"none","encoding":"jsonParsed"}]`, wantRewards: rewards}, + {name: "none-base64-without-rewards", params: `[1,{"transactionDetails":"none","encoding":"base64","rewards":false}]`}, + {name: "none-null-options", params: `[1,{"transactionDetails":"none","encoding":null,"commitment":null,"rewards":null,"maxSupportedTransactionVersion":null}]`, wantRewards: rewards}, + {name: "none-json-version-one", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":1}]`, wantRewards: rewards}, + {name: "none-base64-version-255", params: `[1,{"transactionDetails":"none","encoding":"base64","maxSupportedTransactionVersion":255}]`, wantRewards: rewards}, + {name: "omitted-config", params: `[1]`, wantError: fullUnavailable}, + {name: "null-config", params: `[1,null]`, wantError: fullUnavailable}, + {name: "empty-config", params: `[1,{}]`, wantError: fullUnavailable}, + {name: "omitted-details", params: `[1,{"encoding":"json","rewards":false}]`, wantError: fullUnavailable}, + {name: "null-details", params: `[1,{"transactionDetails":null}]`, wantError: fullUnavailable}, + {name: "legacy-encoding-config", params: `[1,"base64"]`, wantError: fullUnavailable}, + {name: "unsupported-signatures", params: `[1,{"transactionDetails":"signatures"}]`, wantInvalid: true}, + {name: "unsupported-accounts", params: `[1,{"transactionDetails":"accounts"}]`, wantInvalid: true}, + {name: "unknown-details", params: `[1,{"transactionDetails":"partial"}]`, wantInvalid: true}, + {name: "invalid-details-type", params: `[1,{"transactionDetails":1}]`, wantInvalid: true}, + {name: "unknown-encoding", params: `[1,{"transactionDetails":"none","encoding":"base64+zstd"}]`, wantInvalid: true}, + {name: "invalid-encoding-type", params: `[1,{"transactionDetails":"none","encoding":false}]`, wantInvalid: true}, + {name: "invalid-config-type", params: `[1,[]]`, wantInvalid: true}, + {name: "processed-commitment", params: `[1,{"transactionDetails":"none","commitment":"processed"}]`, wantInvalid: true}, + {name: "negative-version", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":-1}]`, wantInvalid: true}, + {name: "overflow-version", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":256}]`, wantInvalid: true}, + {name: "fractional-version", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":0.5}]`, wantInvalid: true}, + {name: "invalid-version-type", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":"0"}]`, wantInvalid: true}, + {name: "rounded-fractional-version", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":0.99999999999999999999}]`, wantInvalid: true}, + {name: "decimal-integer-version", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":1.0}]`, wantInvalid: true}, + {name: "exponent-integer-version", params: `[1,{"transactionDetails":"none","maxSupportedTransactionVersion":1e0}]`, wantInvalid: true}, + {name: "rounded-fractional-slot", params: `[0.99999999999999999999,{"transactionDetails":"none"}]`, wantInvalid: true}, + {name: "decimal-integer-slot", params: `[1.0,{"transactionDetails":"none"}]`, wantInvalid: true}, + {name: "exponent-integer-slot", params: `[1e0,{"transactionDetails":"none"}]`, wantInvalid: true}, + {name: "null-slot", params: `[null,{"transactionDetails":"none"}]`, wantInvalid: true}, + {name: "negative-slot", params: `[-1,{"transactionDetails":"none"}]`, wantInvalid: true}, + {name: "overflow-slot", params: `[18446744073709551616,{"transactionDetails":"none"}]`, wantInvalid: true}, + {name: "exact-large-slot", params: `[9007199254740993,{"transactionDetails":"none"}]`, wantCode: -32004, wantError: "Block not available for slot 9007199254740993"}, + {name: "max-uint64-slot", params: `[18446744073709551615,{"transactionDetails":"none"}]`, wantCode: -32004, wantError: "Block not available for slot 18446744073709551615"}, + {name: "base58-version-one", params: `[1,{"transactionDetails":"none","encoding":"base58","maxSupportedTransactionVersion":1}]`, wantError: "base58 encoding is not supported with maxSupportedTransactionVersion >= 1"}, + {name: "binary-version-one", params: `[1,{"transactionDetails":"none","encoding":"binary","maxSupportedTransactionVersion":1}]`, wantError: "base58 encoding is not supported with maxSupportedTransactionVersion >= 1"}, + } { + t.Run(test.name, func(t *testing.T) { + request := `{"jsonrpc":"2.0","id":7,"method":"getBlock","params":` + test.params + `}` + response, err := httpServer.Client().Post(httpServer.URL, "application/json", strings.NewReader(request)) + require.NoError(t, err) + defer response.Body.Close() + var reply struct { + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` + Result map[string]json.RawMessage `json:"result"` + Error *rpcProbeError `json:"error"` + } + require.NoError(t, json.NewDecoder(response.Body).Decode(&reply)) + require.Equal(t, "2.0", reply.JSONRPC) + require.Equal(t, 7, reply.ID) + if test.wantError != "" || test.wantInvalid { + require.NotNil(t, reply.Error) + wantCode := test.wantCode + if wantCode == 0 { + wantCode = -32602 + } + require.Equal(t, wantCode, reply.Error.Code) + if test.wantError != "" { + require.Equal(t, test.wantError, reply.Error.Message) + } + require.Nil(t, reply.Result) + return + } + require.Equal(t, http.StatusOK, response.StatusCode) + require.Nil(t, reply.Error) + if test.wantRewards == "" { + require.NotContains(t, reply.Result, "rewards") + } else { + require.JSONEq(t, test.wantRewards, string(reply.Result["rewards"])) + } + delete(reply.Result, "rewards") + require.Equal(t, map[string]json.RawMessage{ + "blockHeight": json.RawMessage(`30`), "blockTime": json.RawMessage(`null`), + "blockhash": json.RawMessage(`"11111111111111111111111111111111"`), + "previousBlockhash": json.RawMessage(`"11111111111111111111111111111111"`), + "parentSlot": json.RawMessage(`0`), + }, reply.Result) + }) + } +}