feat(staking,poll): incremental voter weight view + snapshot at PutPollResult (IIP-59 PR 2/6)

Second PR of the IIP-59 protocol-native voter reward distribution
series (spec: iotexproject/iips#73, previous:
PR 1 — commissionRate schema + SetCommissionRate action).

Adds the per-epoch freeze that PR 3's GrantEpochReward will consume:
at PutPollResult (mid-epoch snapshot of the next-epoch active
delegate set), we now capture (a) the delegate's currently-set
commission rate and (b) the per-voter aggregated vote weight for
every delegate that will be paid next epoch. No reward-distribution
logic yet — GrantEpochReward still runs the pre-IIP-59 path; the
snapshot just sits in state until PR 3 reads it. Behavior is
unchanged pre-flag.

Design change vs. the earlier draft on this branch
--------------------------------------------------

The prior version of this PR scanned all buckets at PutPollResult
time to build the per-voter aggregate. That was O(all buckets)
inside a hot state.db call and, worse, required reading from the
contract staking indexer — a db independent of state.db — inside a
consensus mutation path. Both problems are gone in this redesign:

- A live in-memory VoterWeightView is maintained incrementally by
  the native handlers and by the contract-staking receipt-driven
  event dispatch. PutPollResult reads directly from the view (top-N
  candidates only, O(voters-per-cand)), sorts by voter address
  bytes, and writes a per-candidate blob.
- The consensus path (Handle → contractsStake.Handle) never
  touches the contract staking indexer db. The view is fed through
  the same receipt-driven dispatch that today updates
  contractStakeView, via a `VoterDeltaSink` plumbed through
  context (no widening of ContractStakeView interface, no
  breakage for existing mocks/impls).
- The view is memory-only. On protocol Start, `CreateBaseView`
  rebuilds it once by scanning current native buckets and asking
  each contract-staking indexer for its current per-voter
  aggregates. The durable half of the story is the frozen
  per-epoch snapshot blobs written under `_voterWeightSnap`, not
  the live view.

Storage layout
--------------

- New 1-byte namespace tag `_voterWeightSnap = 5` in the staking
  namespace (protocol.go), following the existing _const / _bucket
  / _voterIndex / _candIndex / _endorsement pattern.
- Per-candidate blob key = `_voterWeightSnap || candIdentifier.Bytes()`
  → 21 bytes, mirroring `_voterIndex` / `_candIndex` layout.
- Blob payload = `stakingpb.VoterWeightSnapshot { repeated
  VoterWeightEntry (bytes voter, bytes weight) }`. Entries are
  written pre-sorted by voter address bytes; the marshalled output
  is byte-deterministic given identical logical state — a load-
  bearing invariant for the byte-equality skip in the writer and
  for cross-node consensus.

Live view
---------

- `action/protocol/staking/voter_weight_view.go`: VoterWeightView
  interface + baseVoterWeightView (map[candID]map[voterID]weight)
  + wrapVoterWeightView (change overlay). Wrap layers a change map
  on top of the receiver so Snapshot returns cheaply and Revert
  discards the overlay. Fork deep-clones for parallel working
  sets. Commit folds change into base and prunes zero-weight
  entries.
- Deterministic ordering: `Weights(cand)` returns
  `[]VoterWeight` sorted by voter address bytes. No map iteration
  reaches the consensus-visible output.
- Nil / zero delta is a no-op; negative deltas model bucket exit
  (unstake, withdraw, ChangeCandidate off-ramp, TransferStake off-
  ramp).

Native handler hooks
--------------------

Every native handler that changes a voter bucket's contribution to
a candidate calls `csm.ApplyVoterWeightDelta(cand, voter, Δ)`:

- handleCreateStake: +weightedVote
- handleUnstake: -weightedVote (skipped for self-stake buckets)
- handleChangeCandidate: -w on prev, +w on new (skipped when the
  bucket was prev's self-stake bucket — it was never in the view)
- handleTransferStake: -w on old voter, +w on new voter (skipped
  when bucket is already unstaked — it exited the view earlier)
- handleDepositToStake / handleRestake: +Δw on the same
  (cand, voter) pair (skipped for self-stake buckets)
- handleCandidateEndorsement: no-op on the voter view (self-stake
  gate flips SelfStakeBucketIdx but does not add/remove voter
  buckets)
- handleStakeMigrate: -native_w on (cand, oldOwner); the new
  contract bucket enters the view via the contract-staking event
  sink below

Self-stake buckets are never in the view. The exclusion rule is
uniform: `ContractAddress == "" && Index == cand.SelfStakeBucketIdx`
— fixes review issue #5 on the PoC PR (#4811) where any Index=0
contract bucket was silently treated as self-stake.

Contract staking sink
---------------------

- `staking/protocol.go` wraps `contractsStake.Handle` with
  `WithVoterDeltaSink(ctx, viewData.voterWeights)`. All V1/V2/V3
  event processors get the sink through context.
- `systemcontractindex/stakingindex/vote_view_handler.go` pulls
  the sink out of context and emits deltas on PutBucket/DeleteBucket:
  fresh puts emit +w; owner or candidate changes emit split (-old,
  +new); deletes emit -w. Muted buckets emit no delta (matching
  the "not in view" invariant).
- All sink calls funnel through `VoterWeightView.Apply` which is
  the single source of truth for the aggregation.

Snapshot entry point (staking)
------------------------------

- `action/protocol/staking/voter_weight_snapshot.go`:
  `Protocol.SnapshotForEpochReward(ctx, sm, cands)` is the public
  API poll calls. For each candidate in the next-epoch active list:
  1. copies `staking.Candidate.CommissionRate` into
     `state.Candidate.CommissionRate`. poll then persists this via
     its existing `NxtCandidateKey` PutState — no extra state key
     needed, the `shiftCandidates` path already promotes next →
     current at the epoch boundary.
  2. reads `vd.voterWeights.Weights(candID)` (already sorted),
     encodes with `encodeVoterWeightSnapshot`, and writes an
     incremental per-candidate blob under `_voterWeightSnap` —
     with a byte-equality skip against the existing stored blob
     so steady-state (voter set unchanged epoch-over-epoch) does
     zero writes.
- Guarded on `FeatureCtx.NoVoterRewardDistribution` — a no-op
  pre-fork.
- Empty entries after aggregation → `DelState`, not "write empty
  blob." `VoterWeightsFromSnapshot` returns (nil, nil) for the
  ErrStateNotExist path, so PR 3 sees the same result either way,
  but keeping state clean matches how the other candidate-scoped
  namespaces behave when nothing is left to store.
- State read errors other than ErrStateNotExist propagate wrapped.
  Fixes review issue #4 on the PoC — corruption / IO errors no
  longer silently drop voter entries.

Reader for PR 3
---------------

- `VoterWeightsFromSnapshot(sr, candID) → ([]VoterWeight, error)`:
  a single sm.State call, decoded from the persisted blob.
  (nil, nil) for a candidate with no snapshot (either not yet
  activated, or all voters left).

Tests
-----

- `voter_weight_view_test.go` (19 cases): base apply/aggregate/read,
  sorting stability, wrap overlay semantics, Snapshot/Revert parity
  against a Fork+Apply reference, Commit fold + zero-prune, nested
  Wrap, per-candidate isolation, sink context plumbing.
- `vote_view_handler_test.go` (9 cases, systemcontractindex/
  stakingindex): PutBucket fresh / same-cand-same-owner /
  owner-change / candidate-change; DeleteBucket / missing-bucket
  / muted-bucket; nil-sink tolerance; sink integration that a
  fresh-then-delete sequence balances to zero after Commit.
- `voter_weight_snapshot_test.go`: updated to construct a
  populated view rather than seeding buckets, then assert
  SnapshotForEpochReward writes the expected blob and
  byte-equality skip triggers on unchanged inputs.

Author: envestcc

action/protocol/poll/util.go

@@ -21,6 +21,7 @@ import (
 	"github.com/iotexproject/iotex-core/v2/action/protocol"
 	accountutil "github.com/iotexproject/iotex-core/v2/action/protocol/account/util"
 	"github.com/iotexproject/iotex-core/v2/action/protocol/rolldpos"
+	"github.com/iotexproject/iotex-core/v2/action/protocol/staking"
 	"github.com/iotexproject/iotex-core/v2/action/protocol/vote"
 	"github.com/iotexproject/iotex-core/v2/action/protocol/vote/candidatesutil"
 	"github.com/iotexproject/iotex-core/v2/pkg/log"
@@ -212,6 +213,16 @@ func setCandidates(
 			return errors.Wrapf(err, "failed to put candidatelist into indexer at height %d", height)
 		}
 	}
+	// IIP-59: freeze per-epoch commission rate + per-voter weights for the
+	// next epoch's active delegates. No-op when the feature is disabled.
+	// Must run before PutState below — the commission-rate copy mutates
+	// candidates in place, and the persisted state.Candidate needs the
+	// frozen value.
+	if stakingProto := staking.FindProtocol(protocol.MustGetRegistry(ctx)); stakingProto != nil {
+		if err := stakingProto.SnapshotForEpochReward(ctx, sm, candidates); err != nil {
+			return errors.Wrap(err, "failed to snapshot voter reward state for next epoch")
+		}
+	}
 	if loadCandidatesLegacy {
 		key, org := candidatesutil.ConstructLegacyKeyWithOrg(height)
 		_, err := sm.PutState(&candidates, protocol.LegacyKeyOption(key), protocol.ErigonStoreKeyOption(org))

action/protocol/staking/candidate_statemanager.go

@@ -57,6 +57,11 @@ type (
 		Upsert(*Candidate) error
 		CreditBucketPool(*big.Int, bool) error
 		DebitBucketPool(*big.Int, bool) error
+		// ApplyVoterWeightDelta pushes a (cand, voter, Δweight) delta into
+		// the incremental voter-weight view. No-op when IIP-59 voter reward
+		// distribution is inactive (pre-fork or view not yet initialized).
+		// Handlers call this at each SubVote / AddVote site.
+		ApplyVoterWeightDelta(cand, voter address.Address, delta *big.Int)
 		Commit(context.Context) error
 		SM() protocol.StateManager
 		SR() protocol.StateReader
@@ -124,6 +129,7 @@ func (csm *candSM) DirtyView() *viewData {
 		candCenter:     csm.candCenter,
 		bucketPool:     csm.bucketPool,
 		contractsStake: vd.contractsStake,
+		voterWeights:   vd.voterWeights,
 	}
 }
 
@@ -176,6 +182,23 @@ func (csm *candSM) CreditBucketPool(amount *big.Int, deleteBucket bool) error {
 	return csm.bucketPool.CreditPool(csm.StateManager, amount, deleteBucket)
 }
 
+// ApplyVoterWeightDelta forwards to the live voterWeights view stored on the
+// protocol viewData. It reads the view fresh each call — Snapshot()/Revert()
+// swap the view instance in place, so a cached pointer would go stale after
+// a mid-action snapshot. When the view is nil (pre-fork or no IIP-59
+// initialization), the call is a no-op so handlers can invoke it
+// unconditionally without extra feature-flag gating.
+func (csm *candSM) ApplyVoterWeightDelta(cand, voter address.Address, delta *big.Int) {
+	if delta == nil || delta.Sign() == 0 {
+		return
+	}
+	view := csm.DirtyView()
+	if view == nil || view.voterWeights == nil {
+		return
+	}
+	view.voterWeights.Apply(cand, voter, delta)
+}
+
 func (csm *candSM) DebitBucketPool(amount *big.Int, newBucket bool) error {
 	return csm.bucketPool.DebitPool(csm, amount, newBucket)
 }

action/protocol/staking/candidate_statereader.go

@@ -141,6 +141,7 @@ func ConstructBaseView(sr protocol.StateReader) (CandidateStateReader, error) {
 			candCenter:     view.candCenter,
 			bucketPool:     view.bucketPool,
 			contractsStake: view.contractsStake,
+			voterWeights:   view.voterWeights,
 		},
 	}, nil
 }

action/protocol/staking/handler_candidate_endorsement.go

@@ -84,6 +84,7 @@ func (p *Protocol) handleCandidateEndorsement(ctx context.Context, act *action.C
 				if err := csm.deactivate(cand, bucket, protocol.MustGetBlockCtx(ctx).BlockHeight, p.calculateVoteWeight); err != nil {
 					return log, nil, csmErrorToHandleError(cand.GetIdentifier().String(), err)
 				}
+				csm.ApplyVoterWeightDelta(cand.GetIdentifier(), bucket.Owner, p.calculateVoteWeight(bucket, false))
 			} else {
 				// TODO: Check that the bucket is ready for dequeue
 				if err := p.clearCandidateSelfStake(bucket, cand); err != nil {
@@ -92,6 +93,7 @@ func (p *Protocol) handleCandidateEndorsement(ctx context.Context, act *action.C
 				if err := csm.Upsert(cand); err != nil {
 					return log, nil, csmErrorToHandleError(actCtx.Caller.String(), err)
 				}
+				csm.ApplyVoterWeightDelta(cand.GetIdentifier(), bucket.Owner, p.calculateVoteWeight(bucket, false))
 			}
 		}
 		if err := esm.Delete(bucket.Index); err != nil {

action/protocol/staking/handler_stake_migrate.go

@@ -129,6 +129,7 @@ func (p *Protocol) withdrawBucket(ctx context.Context, withdrawer *state.Account
 		return nil, nil, errors.Wrapf(err, "failed to update staking bucket pool %s", err.Error())
 	}
 	// update candidate vote
+	wasSelfStake := cand.SelfStakeBucketIdx == bucket.Index
 	weightedVote := p.calculateVoteWeight(bucket, false)
 	if err := cand.SubVote(weightedVote); err != nil {
 		return nil, nil, &handleError{
@@ -137,13 +138,16 @@ func (p *Protocol) withdrawBucket(ctx context.Context, withdrawer *state.Account
 		}
 	}
 	// clear candidate's self stake if the
-	if cand.SelfStakeBucketIdx == bucket.Index {
+	if wasSelfStake {
 		cand.SelfStake = big.NewInt(0)
 		cand.SelfStakeBucketIdx = candidateNoSelfStakeBucketIndex
 	}
 	if err := csm.Upsert(cand); err != nil {
 		return nil, nil, csmErrorToHandleError(cand.GetIdentifier().String(), err)
 	}
+	if !wasSelfStake {
+		csm.ApplyVoterWeightDelta(cand.GetIdentifier(), bucket.Owner, new(big.Int).Neg(weightedVote))
+	}
 	// update withdrawer balance
 	if err := withdrawer.AddBalance(bucket.StakedAmount); err != nil {
 		return nil, nil, errors.Wrapf(err, "failed to add balance %s", bucket.StakedAmount)

action/protocol/staking/handlers.go

@@ -91,6 +91,8 @@ func (p *Protocol) handleCreateStake(ctx context.Context, act *action.CreateStak
 			failureStatus: iotextypes.ReceiptStatus_ErrInvalidBucketAmount,
 		}
 	}
+	// IIP-59: new voter bucket, credit the voter's weight to this candidate.
+	csm.ApplyVoterWeightDelta(candidate.GetIdentifier(), bucket.Owner, weightedVote)
 	if err := csm.Upsert(candidate); err != nil {
 		return log, nil, csmErrorToHandleError(candidate.GetIdentifier().String(), err)
 	}
@@ -213,6 +215,11 @@ func (p *Protocol) handleUnstake(ctx context.Context, act *action.Unstake, csm C
 			failureStatus: iotextypes.ReceiptStatus_ErrNotEnoughBalance,
 		}
 	}
+	// IIP-59: voter buckets exit the view on unstake. Self-stake buckets are
+	// never in the view — skip the delta for those.
+	if !selfStake {
+		csm.ApplyVoterWeightDelta(candidate.GetIdentifier(), bucket.Owner, new(big.Int).Neg(weightedVote))
+	}
 	// clear candidate's self stake if the bucket is self staking
 	if selfStake {
 		candidate.SelfStake = big.NewInt(0)
@@ -369,15 +376,22 @@ func (p *Protocol) handleChangeCandidate(ctx context.Context, act *action.Change
 
 	// update previous candidate
 	weightedVotes := p.calculateVoteWeight(bucket, false)
+	// IIP-59: capture pre-write self-stake state so we can correctly gate the
+	// voter-weight delta on prev. If this bucket was prev's self-stake bucket
+	// it was NOT in the view (self-stake exclusion), so we must skip the -w.
+	wasSelfStakeOnPrev := !featureCtx.DisableDelegateEndorsement && prevCandidate.SelfStakeBucketIdx == bucket.Index
 	if err := prevCandidate.SubVote(weightedVotes); err != nil {
 		return log, &handleError{
 			err:           errors.Wrapf(err, "failed to subtract vote for previous candidate %s", prevCandidate.GetIdentifier().String()),
 			failureStatus: iotextypes.ReceiptStatus_ErrNotEnoughBalance,
 		}
 	}
+	if !wasSelfStakeOnPrev {
+		csm.ApplyVoterWeightDelta(prevCandidate.GetIdentifier(), bucket.Owner, new(big.Int).Neg(weightedVotes))
+	}
 	// if the bucket equals to the previous candidate's self-stake bucket, it must be expired endorse bucket
 	// so we need to clear the self-stake of the previous candidate
-	if !featureCtx.DisableDelegateEndorsement && prevCandidate.SelfStakeBucketIdx == bucket.Index {
+	if wasSelfStakeOnPrev {
 		prevCandidate.SelfStake.SetInt64(0)
 		prevCandidate.SelfStakeBucketIdx = candidateNoSelfStakeBucketIndex
 	}
@@ -392,6 +406,9 @@ func (p *Protocol) handleChangeCandidate(ctx context.Context, act *action.Change
 			failureStatus: iotextypes.ReceiptStatus_ErrInvalidBucketAmount,
 		}
 	}
+	// IIP-59: bucket now votes for the new candidate — always a voter contribution,
+	// never self-stake (ChangeCandidate does not create a self-stake link).
+	csm.ApplyVoterWeightDelta(candidate.GetIdentifier(), bucket.Owner, weightedVotes)
 	if err := csm.Upsert(candidate); err != nil {
 		return log, csmErrorToHandleError(candidate.GetIdentifier().String(), err)
 	}
@@ -437,6 +454,15 @@ func (p *Protocol) handleTransferStake(ctx context.Context, act *action.Transfer
 		}
 	}
 
+	// IIP-59: TransferStake reassigns the bucket's voter without touching the
+	// candidate's total. If the bucket was already unstaked it was not in the
+	// view; skip the delta in that case.
+	var voterDelta *big.Int
+	prevOwner := bucket.Owner
+	if !bucket.isUnstaked() {
+		voterDelta = p.calculateVoteWeight(bucket, false)
+	}
+
 	// update bucket index
 	if err := csm.delVoterBucketIndex(bucket.Owner, act.BucketIndex()); err != nil {
 		return log, errors.Wrapf(err, "failed to delete voter bucket index for voter %s", bucket.Owner.String())
@@ -451,6 +477,11 @@ func (p *Protocol) handleTransferStake(ctx context.Context, act *action.Transfer
 		return log, errors.Wrapf(err, "failed to update bucket for voter %s", bucket.Owner.String())
 	}
 
+	if voterDelta != nil && voterDelta.Sign() > 0 {
+		csm.ApplyVoterWeightDelta(bucket.Candidate, prevOwner, new(big.Int).Neg(voterDelta))
+		csm.ApplyVoterWeightDelta(bucket.Candidate, newOwner, voterDelta)
+	}
+
 	log.AddAddress(actionCtx.Caller)
 	return log, nil
 }
@@ -549,6 +580,11 @@ func (p *Protocol) handleDepositToStake(ctx context.Context, act *action.Deposit
 			failureStatus: iotextypes.ReceiptStatus_ErrInvalidBucketAmount,
 		}
 	}
+	// IIP-59: voter contribution only. Self-stake buckets never enter the view.
+	if !selfStake {
+		delta := new(big.Int).Sub(weightedVotes, prevWeightedVotes)
+		csm.ApplyVoterWeightDelta(candidate.GetIdentifier(), bucket.Owner, delta)
+	}
 	if selfStake {
 		if err := candidate.AddSelfStake(act.Amount()); err != nil {
 			return log, nil, &handleError{
@@ -668,6 +704,11 @@ func (p *Protocol) handleRestake(ctx context.Context, act *action.Restake, csm C
 			failureStatus: iotextypes.ReceiptStatus_ErrInvalidBucketAmount,
 		}
 	}
+	// IIP-59: voter contribution only. Self-stake buckets never enter the view.
+	if !selfStake {
+		delta := new(big.Int).Sub(weightedVotes, prevWeightedVotes)
+		csm.ApplyVoterWeightDelta(candidate.GetIdentifier(), bucket.Owner, delta)
+	}
 	if err := csm.Upsert(candidate); err != nil {
 		return log, csmErrorToHandleError(candidate.GetIdentifier().String(), err)
 	}

action/protocol/staking/protocol.go

@@ -61,6 +61,9 @@ const (
 	_voterIndex
 	_candIndex
 	_endorsement
+	// _voterWeightSnap holds IIP-59 per-candidate frozen voter weight blobs,
+	// written at PutPollResult and read by rewarding at GrantEpochReward.
+	_voterWeightSnap
 )
 
 // Errors
@@ -374,9 +377,49 @@ func (p *Protocol) Start(ctx context.Context, sr protocol.StateReader) (protocol
 			return nil, err
 		}
 	}
+
+	// IIP-59: build the initial voter weight base once, from state.db native
+	// buckets + contract staking indexers verified above. Skip the scan when
+	// the fork isn't active — the view stays nil and Apply calls are no-ops.
+	if fCtx := protocol.MustGetFeatureCtx(ctx); !fCtx.NoVoterRewardDistribution {
+		csr := newCandidateStateReader(sr)
+		vw, err := buildVoterWeightBaseFromState(
+			csr,
+			c.candCenter.All(),
+			p.activeContractStakingIndexers(height),
+			height,
+			p.config.VoteWeightCalConsts,
+		)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to build voter weight base view")
+		}
+		c.voterWeights = vw
+	}
 	return c, nil
 }
 
+// activeContractStakingIndexers returns the registered indexers whose
+// StartHeight has already been reached at the given height. Nil entries are
+// omitted. Used at protocol Start for the initial voter-weight base scan.
+func (p *Protocol) activeContractStakingIndexers(height uint64) []ContractStakingIndexer {
+	var out []ContractStakingIndexer
+	candidates := []ContractStakingIndexer{
+		p.contractStakingIndexer,
+		p.contractStakingIndexerV2,
+		p.contractStakingIndexerV3,
+	}
+	for _, idx := range candidates {
+		if idx == nil {
+			continue
+		}
+		if idx.StartHeight() > height {
+			continue
+		}
+		out = append(out, idx)
+	}
+	return out
+}
+
 // CreateGenesisStates is used to setup BootstrapCandidates from genesis config.
 func (p *Protocol) CreateGenesisStates(
 	ctx context.Context,
@@ -837,7 +880,12 @@ func (p *Protocol) HandleReceipt(ctx context.Context, elp action.Envelope, sm pr
 		if err != nil {
 			return err
 		}
-		if err := v.(*viewData).contractsStake.Handle(ctx, receipt); err != nil {
+		vd := v.(*viewData)
+		handleCtx := ctx
+		if vd.voterWeights != nil {
+			handleCtx = WithVoterDeltaSink(ctx, vd.voterWeights)
+		}
+		if err := vd.contractsStake.Handle(handleCtx, receipt); err != nil {
 			return err
 		}
 		handler = newNFTBucketEventHandlerErigonOnly(sm, ccvw)