Add inflation reward RPC support

.github/workflows/go_build.yml

@@ -39,6 +39,7 @@ jobs:
         run: >-
           go test -race -p 2 -count=1
           ./pkg/alpenglow ./pkg/consensus ./pkg/replay ./pkg/rewards
+          ./pkg/rpcserver ./pkg/epochrewards
           ./pkg/turbine ./pkg/sigverify ./pkg/blockprod/...
           ./cmd/mithril/node ./cmd/mithril/configcmd
 

README.md

@@ -222,8 +222,20 @@ curl http://YOUR_MITHRIL_IP:8899 -X POST -H "Content-Type: application/json" -d
 - `getBankHash` - Get bankhash for a slot (Mithril extension, not standard Solana RPC)
 - `getBlockHeight` - Get current block height
 - `getEpochInfo` - Get current epoch info
+- `getInflationReward` - Get retained inflation rewards for an epoch
 - `getLatestBlockhash` - Get recent blockhash
 
+On Alpenglow, `getInflationReward` can publish verified canonical rewards before
+the account checkpoint reaches disk. `confirmed` and `finalized` share that view;
+`processed` is not supported. The default epoch and `minContextSlot` check use the
+same reward-read slot. Other RPC methods may still use an older durable view.
+Recovery withdraws the in-memory view, and a restart restores only
+durable history. Retention follows durable checkpoints. Advisory or disabled
+execution verification keeps reward reads on the durable view.
+Credited rewards are readable while other staking partitions are still pending.
+Requests for pending partitions return `-32017` with verified distribution
+progress; pruned history returns `-32001`, and unknown history returns `-32004`.
+
 We're actively expanding RPC method coverage. Upcoming methods include transaction simulation, send transaction, and get leader schedule.
 
 ### Current Limitations

cmd/mithril/node/epoch_reward_recovery_test.go

@@ -0,0 +1,139 @@
+package node
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/Overclock-Validator/mithril/pkg/accounts"
+	"github.com/Overclock-Validator/mithril/pkg/accountsdb"
+	"github.com/Overclock-Validator/mithril/pkg/block"
+	"github.com/Overclock-Validator/mithril/pkg/rpcserver"
+	"github.com/Overclock-Validator/mithril/pkg/sealevel"
+	"github.com/Overclock-Validator/mithril/pkg/state"
+	"github.com/gagliardetto/solana-go"
+	"github.com/gagliardetto/solana-go/rpc"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEpochRewardHTTPAfterDurableRewind(t *testing.T) {
+	for _, name := range []string{"rewind", "prune failure", "accounts cleanup failure", "accounts precommit failure"} {
+		t.Run(name, func(t *testing.T) {
+			failPrune := name == "prune failure"
+			dir := t.TempDir()
+			require.NoError(t, os.MkdirAll(filepath.Join(dir, "accounts"), 0o755))
+			require.NoError(t, os.WriteFile(filepath.Join(dir, "largest_file_id"), make([]byte, 8), 0o644))
+			require.NoError(t, os.WriteFile(filepath.Join(dir, "bootstrap_high_file_id"), make([]byte, 8), 0o644))
+			db, err := accountsdb.OpenDb(dir)
+			require.NoError(t, err)
+			db.RootedDurable = true
+			db.InitCaches()
+			t.Cleanup(db.CloseDb)
+			address := solana.PublicKey{91}
+			var tip *state.ResumeContext
+			for _, slot := range []uint64{2, 4} {
+				tip = testDurableStatusContext(t, dir, slot)
+				encoded, err := json.Marshal(tip)
+				require.NoError(t, err)
+				_, err = db.CommitBatch([]accounts.SlotDelta{{Slot: slot, Delta: []*accounts.Account{{Key: address, Lamports: slot}}}}, slot, nil, encoded)
+				require.NoError(t, err)
+			}
+			server := rpcserver.NewRpcServer(db, 0, &sealevel.SysvarEpochSchedule{SlotsPerEpoch: 3}, solana.Hash{1})
+			require.NoError(t, server.EnableEpochRewards(filepath.Join(dir, "rpc-epoch-rewards"), 100, 0))
+			record := &block.Block{Slot: 4, Epoch: 1, Rewards: []rpc.BlockReward{{Pubkey: address, Lamports: 2, PostBalance: 4, RewardType: rpc.RewardTypeVoting}}}
+			require.NoError(t, server.RecordEpochRewards(record))
+			require.NoError(t, server.PrepareEpochRewards(4))
+			require.NoError(t, server.SetRootedEpochRewardsSlot(4))
+			endpoint := httptest.NewServer(server)
+			t.Cleanup(endpoint.Close)
+			endpoint.Client().Timeout = time.Second
+			call := func(minContextSlot uint64) map[string]json.RawMessage {
+				t.Helper()
+				body, err := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "getInflationReward", "params": []any{[]string{address.String()}, map[string]any{"epoch": 0, "minContextSlot": minContextSlot}}})
+				require.NoError(t, err)
+				response, err := endpoint.Client().Post(endpoint.URL, "application/json", bytes.NewReader(body))
+				require.NoError(t, err)
+				defer response.Body.Close()
+				payload, err := io.ReadAll(response.Body)
+				require.NoError(t, err)
+				t.Logf("HTTP=%d response=%s", response.StatusCode, payload)
+				var result map[string]json.RawMessage
+				require.NoError(t, json.Unmarshal(payload, &result))
+				return result
+			}
+			want := `[{"epoch":0,"effectiveSlot":4,"amount":2,"postBalance":4,"commission":null}]`
+			unavailable := `{"code":-32004,"message":"Epoch reward history is unavailable for epoch 0","data":{"epoch":0,"contextSlot":2}}`
+			require.JSONEq(t, want, string(call(4)["result"]))
+			// An in-memory fork switch above the durable root must retain its rewards.
+			require.NoError(t, server.RewindEpochRewards(5))
+			require.JSONEq(t, want, string(call(4)["result"]))
+
+			s := &state.MithrilState{LastRootedSlot: 4, LastRootedContext: tip, LastSlot: 4}
+			if failPrune {
+				batchPath := filepath.Join(dir, "rpc-epoch-rewards", "batch-00000000000000000004.json")
+				require.NoError(t, os.Remove(batchPath))
+				require.NoError(t, os.Mkdir(batchPath, 0o755))
+				require.NoError(t, os.WriteFile(filepath.Join(batchPath, "blocker"), nil, 0o644))
+			}
+			var accountFaultPath string
+			if name == "accounts cleanup failure" {
+				accountFaultPath = filepath.Join(db.AcctsDir, "rewound")
+				require.NoError(t, os.WriteFile(accountFaultPath, nil, 0o644))
+			}
+			if name == "accounts precommit failure" {
+				manifests, err := accountsdb.ListFoldManifests(db.AcctsDir)
+				require.NoError(t, err)
+				for _, manifest := range manifests {
+					if manifest.ThroughSlot == 4 {
+						accountFaultPath = manifest.Path + ".rewound"
+					}
+				}
+				require.NotEmpty(t, accountFaultPath)
+				require.NoError(t, os.Mkdir(accountFaultPath, 0o755))
+			}
+			require.Equal(t, name == "rewind", rewindStoreBelowDivergence(dir, db, s, 3, 100, server))
+			account, err := db.GetAccount(2, address)
+			require.NoError(t, err)
+			if name == "accounts precommit failure" {
+				require.Equal(t, uint64(4), account.Lamports, "failed manifest parking must not alter accounts")
+			} else {
+				require.Equal(t, uint64(2), account.Lamports)
+			}
+			if name != "rewind" {
+				for _, minContextSlot := range []uint64{2, 4} {
+					reply := call(minContextSlot)
+					require.Empty(t, reply["result"])
+					require.Contains(t, string(reply["error"]), "RPC history is unavailable during account recovery")
+				}
+				if accountFaultPath == "" {
+					// Inspect the side-store watermark independently of the gate; recovery remains halted in production.
+					server.SetHistoryRecoveryPending(false)
+					require.JSONEq(t, `{"code":-32016,"message":"Minimum context slot has not been reached","data":{"contextSlot":2}}`, string(call(4)["error"]))
+					require.JSONEq(t, unavailable, string(call(2)["error"]))
+					return
+				}
+				require.FileExists(t, filepath.Join(dir, "rpc-epoch-rewards", "batch-00000000000000000004.json"), "failed account rewind must retain rewards")
+				require.NoError(t, os.Remove(accountFaultPath))
+				require.True(t, rewindStoreBelowDivergence(dir, db, s, 3, 100, server))
+			}
+			require.Equal(t, uint64(2), s.LastRootedSlot)
+			// Replay's startup hook must not restore the discarded durable reward.
+			require.NoError(t, server.RewindEpochRewards(s.GetResumeSlot()))
+			require.JSONEq(t, `{"code":-32016,"message":"Minimum context slot has not been reached","data":{"contextSlot":2}}`, string(call(4)["error"]))
+			require.JSONEq(t, unavailable, string(call(2)["error"]))
+
+			// A replacement on the selected fork becomes visible only after its fold.
+			record.Rewards[0].Lamports, record.Rewards[0].PostBalance = 3, 5
+			require.NoError(t, server.RecordEpochRewards(record))
+			require.NoError(t, server.PrepareEpochRewards(4))
+			require.JSONEq(t, unavailable, string(call(2)["error"]))
+			require.NoError(t, server.SetRootedEpochRewardsSlot(4))
+			require.JSONEq(t, `[{"epoch":0,"effectiveSlot":4,"amount":3,"postBalance":5,"commission":null}]`, string(call(4)["result"]))
+		})
+	}
+}

cmd/mithril/node/node.go

@@ -2574,7 +2574,16 @@ postBootstrap:
 	if rpcPort < 0 || rpcPort > 65535 {
 		klog.Fatalf("invalid port: %d", rpcPort)
 	} else if rpcPort != 0 {
-		rpcServer = rpcserver.NewRpcServer(accountsDb, uint16(rpcPort), epochScheduleFromState(mithrilState), solana.MustHashFromBase58(networkGenesisHash))
+		schedule := epochScheduleFromState(mithrilState)
+		rpcServer = rpcserver.NewRpcServer(accountsDb, uint16(rpcPort), schedule, solana.MustHashFromBase58(networkGenesisHash))
+		if alpenglowMode {
+			if schedule == nil || schedule.SlotsPerEpoch == 0 || schedule.SlotsPerEpoch > math.MaxUint64/3 {
+				klog.Fatalf("Alpenglow RPC requires a valid epoch schedule")
+			}
+			if err := rpcServer.EnableEpochRewards(filepath.Join(accountsPath, "rpc-epoch-rewards"), 3*schedule.SlotsPerEpoch, mithrilState.LastRootedSlot); err != nil {
+				klog.Fatalf("unable to load RPC epoch rewards: %v", err)
+			}
+		}
 		rpcServer.Start()
 		mlog.Log.Infof("Started RPC server on port %d", rpcPort)
 	}
@@ -4342,7 +4351,11 @@ func completeInterruptedRewind(accountsDbPath string, accountsDb *accountsdb.Acc
 // boundary strictly below divSlot and adopts that boundary's context as the
 // rooted checkpoint. Returns false (caller halts) when no boundary below the
 // divergence is retained or the rewind/reconcile fails.
-func rewindStoreBelowDivergence(accountsDbPath string, accountsDb *accountsdb.AccountsDb, s *state.MithrilState, divSlot uint64, rewindHorizon uint64) bool {
+func rewindStoreBelowDivergence(accountsDbPath string, accountsDb *accountsdb.AccountsDb, s *state.MithrilState, divSlot uint64, rewindHorizon uint64, rpcServer replay.SlotCtxSetter) bool {
+	historyGate, hasHistoryGate := rpcServer.(interface{ SetHistoryRecoveryPending(bool) })
+	if hasHistoryGate {
+		historyGate.SetHistoryRecoveryPending(true)
+	}
 	points, err := accountsDb.ListRewindPoints()
 	if err != nil {
 		mlog.Log.Errorf("fork switch: cannot list rewind boundaries: %v; halting", err)
@@ -4373,12 +4386,21 @@ func rewindStoreBelowDivergence(accountsDbPath string, accountsDb *accountsdb.Ac
 		mlog.Log.Errorf("fork switch: rewind to boundary %d failed: %v; halting", target, err)
 		return false
 	}
+	if rewards, ok := rpcServer.(interface{ SetRootedEpochRewardsSlot(uint64) error }); ok {
+		if err := rewards.SetRootedEpochRewardsSlot(res.NewThrough); err != nil {
+			mlog.Log.Errorf("fork switch: rewind RPC epoch rewards to slot %d: %v; halting", res.NewThrough, err)
+			return false
+		}
+	}
 	if err := adoptRewindResult(accountsDbPath, s, res); err != nil {
 		mlog.Log.Errorf("fork switch: %v; halting", err)
 		return false
 	}
 	mlog.Log.Warnf("fork switch: superseded slot %d was already folded (R=%d); rewound durable state to boundary %d (%d batches / %d keys undone)",
 		divSlot, oldRooted, res.NewThrough, res.UndoneBatches, res.UndoneKeys)
+	if hasHistoryGate {
+		historyGate.SetHistoryRecoveryPending(false)
+	}
 	return true
 }
 
@@ -4607,7 +4629,7 @@ func runReplayWithRecovery(
 		// (this is exactly why durable state deliberately lags and undo pointers
 		// are retained for a horizon). Beyond the horizon -> fail closed.
 		if divSlot <= mithrilState.LastRootedSlot {
-			if !rewindStoreBelowDivergence(accountsDbPath, accountsDb, mithrilState, divSlot, rewindHorizon) {
+			if !rewindStoreBelowDivergence(accountsDbPath, accountsDb, mithrilState, divSlot, rewindHorizon, rpcServer) {
 				break
 			}
 		}

pkg/blockprod/scheduler/scheduler_test.go

@@ -233,6 +233,8 @@ func TestSchedulerSkipsWhenSlotEntryBytesExceeded(t *testing.T) {
 	for sched.Stats().DroppedCost == 0 && time.Now().Before(deadline) {
 		time.Sleep(5 * time.Millisecond)
 	}
+	// Wait for the drain loop to return any temporarily popped entries.
+	sched.Stop()
 	require.Equal(t, uint64(1), sched.Stats().DroppedCost)
 	require.Equal(t, uint64(1), sched.Stats().DroppedBatchBytes)
 	require.Equal(t, uint64(0), sched.Stats().Accepted)

pkg/epochrewards/coverage_test.go

@@ -0,0 +1,109 @@
+package epochrewards
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestEpochCoverageEmptyDistributionAndRetention(t *testing.T) {
+	dir := t.TempDir()
+	store, err := Open(dir, 128)
+	require.NoError(t, err)
+	store.RecordCoverage(0, 128, true, false)
+	require.NoError(t, store.Prepare(128))
+	require.NoError(t, store.SetRooted(128))
+	require.False(t, store.EpochAvailable(0), "an active distribution is incomplete")
+	store.RecordCoverage(0, 129, false, true)
+	require.NoError(t, store.Prepare(129))
+	require.False(t, store.EpochAvailable(0), "prepared completion cannot precede account commit")
+	require.NoError(t, store.SetRooted(129))
+	require.True(t, store.EpochAvailable(0), "a complete epoch may have no payouts")
+	reopened, err := Open(dir, 128)
+	require.NoError(t, err)
+	require.NoError(t, reopened.SetRooted(129))
+	require.True(t, reopened.EpochAvailable(0))
+	require.NoError(t, reopened.SetRooted(256))
+	require.False(t, reopened.EpochAvailable(0), "a retained completion cannot replace a pruned boundary")
+	_, err = os.Stat(reopened.batchPath(129))
+	require.NoError(t, err, "the completion batch itself remains retained")
+}
+
+func TestEpochCoverageRewindAndPartialFoldRecovery(t *testing.T) {
+	dir := t.TempDir()
+	store, err := Open(dir, 512)
+	require.NoError(t, err)
+	store.RecordCoverage(0, 128, true, false)
+	require.NoError(t, store.Prepare(128))
+	store.RecordCoverage(0, 129, false, true)
+	require.NoError(t, store.Prepare(130))
+
+	// Only the first account fold committed. The later duplicate boundary and
+	// completion must not hide or replace the committed boundary on reopen.
+	store, err = Open(dir, 512)
+	require.NoError(t, err)
+	require.NoError(t, store.SetRooted(128))
+	require.False(t, store.EpochAvailable(0))
+	store.RecordCoverage(0, 131, false, true)
+	require.NoError(t, store.Prepare(132))
+	require.NoError(t, store.Rewind(131))
+	require.False(t, store.EpochAvailable(0))
+	store.RecordCoverage(0, 133, false, true)
+	require.NoError(t, store.Prepare(134))
+	require.NoError(t, store.SetRooted(134))
+	require.True(t, store.EpochAvailable(0))
+	store, err = Open(dir, 512)
+	require.NoError(t, err)
+	require.NoError(t, store.SetRooted(134))
+	require.True(t, store.EpochAvailable(0))
+}
+
+func TestEpochCoverageExpiresEvenWhenBothMarkersRemain(t *testing.T) {
+	store, err := Open(t.TempDir(), 128)
+	require.NoError(t, err)
+	store.RecordCoverage(0, 128, true, false)
+	store.RecordCoverage(0, 140, false, true)
+	require.NoError(t, store.Prepare(200))
+	require.NoError(t, store.SetRooted(200))
+	require.True(t, store.EpochAvailable(0))
+	require.NoError(t, store.SetRooted(256))
+	_, err = os.Stat(store.batchPath(200))
+	require.NoError(t, err)
+	require.False(t, store.EpochAvailable(0), "coverage cannot extend beyond the epoch's retained boundary")
+}
+
+func TestEpochCoverageDoesNotInferSnapshotHistory(t *testing.T) {
+	store, err := Open(t.TempDir(), 512)
+	require.NoError(t, err)
+	store.RecordCoverage(0, 150, false, true)
+	require.NoError(t, store.Prepare(150))
+	require.NoError(t, store.SetRooted(150))
+	require.False(t, store.EpochAvailable(0), "inactive rewards after a snapshot do not prove the missing boundary")
+	files, err := batchFiles(store.dir)
+	require.NoError(t, err)
+	require.Empty(t, files)
+
+	// Even a surviving completion event in an older batch is insufficient.
+	require.NoError(t, os.WriteFile(store.batchPath(150), []byte(`{"version":1,"through":150,"records":[],"coverage":[{"epoch":0,"slot":150,"boundarySlot":128,"complete":true}]}`), 0o600))
+	store, err = Open(store.dir, 512)
+	require.NoError(t, err)
+	require.NoError(t, store.SetRooted(150))
+	require.False(t, store.EpochAvailable(0))
+}
+
+func TestEpochCoverageRejectsMalformedBatches(t *testing.T) {
+	for _, coverage := range []string{
+		`[{"epoch":0,"slot":151,"boundarySlot":128,"complete":true}]`,
+		`[{"epoch":0,"slot":150,"boundarySlot":151,"complete":true}]`,
+		`[{"epoch":0,"slot":150,"boundarySlot":128,"complete":false}]`,
+		`[{"epoch":0,"slot":128,"boundarySlot":128},{"epoch":0,"slot":129,"boundarySlot":129}]`,
+	} {
+		dir := t.TempDir()
+		store, err := Open(dir, 512)
+		require.NoError(t, err)
+		require.NoError(t, os.WriteFile(store.batchPath(150), []byte(`{"version":1,"through":150,"records":[],"coverage":`+coverage+`}`), 0o600))
+		_, err = Open(dir, 512)
+		require.ErrorContains(t, err, "invalid epoch reward coverage")
+	}
+}