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