refactor: add a new buffer blocks param to delay deletion of pending ballots (#3553)

* add buffer blocks for pending ballots

* add comments and refactor BuildRewardsDistribution function.

* add changelog

* update comments

* update comments

* update comments

* explicitly set updated params

* add additional checks for migration

* refactor to single ClearFinalizedMaturedBallots with forceDeleteAllBallots flag

* remove SecondsToBlock function

* add pendingBufferBlocks to proposal

* format files

Author: kingpinXD

changelog.md

@@ -33,6 +33,7 @@
 * [3381](https://github.com/zeta-chain/node/pull/3381) - split Bitcoin observer and signer into small files and organize outbound logic into reusable/testable functions; renaming, type unification, etc.
 * [3496](https://github.com/zeta-chain/node/pull/3496) - zetaclient uses `ConfirmationParams` instead of old `ConfirmationCount`; use block ranged based observation for btc and evm chain.
 * [3594](https://github.com/zeta-chain/node/pull/3594) - set outbound hash in cctx when adding outbound tracker
+* [3553](https://github.com/zeta-chain/node/pull/3553) — add a new buffer blocks param to delay deletion of pending ballots
 
 ### Fixes
 

contrib/localnet/orchestrator/proposals_e2e_start/emissions_params.json

@@ -8,13 +8,15 @@
         "tssSignerEmissionPercentage": "0.25",
         "observerSlashAmount": "100000000000000000",
         "ballotMaturityBlocks": 101,
-        "blockRewardAmount": "9620949074074074074.074070733466756687"
+        "blockRewardAmount": "9620949074074074074.074070733466756687",
+        "pendingBallotsDeletionBufferBlocks": 145222
       },
       "authority": "zeta10d07y265gmmuvt4z0w9aw880jnsr700jvxasvr"
     }
   ],
   "metadata": "test proposal",
   "deposit": "100000000azeta",
   "title": "emissions param change",
-  "summary": "emissions param change"
+  "summary": "emissions param change",
+  "min_version": "v29.0.0"
 }

docs/openapi/openapi.swagger.yaml

@@ -58146,6 +58146,9 @@ definitions:
         format: int64
       blockRewardAmount:
         type: string
+      pendingBallotsDeletionBufferBlocks:
+        type: string
+        format: int64
     title: |-
       Params defines the parameters for the module.
       Sample values:
@@ -58155,6 +58158,7 @@ definitions:
          ObserverSlashAmount:         100000000000000000,
          BallotMaturityBlocks:        100,
          BlockRewardAmount:           9620949074074074074.074070733466756687,
+         PendingBallotsDeletionBufferBlocks: 144000
   zetachain.zetacore.emissions.QueryListPoolAddressesResponse:
     type: object
     properties:

e2e/runner/gov_proposals.go

@@ -15,6 +15,7 @@ import (
 	govv1 "github.com/cosmos/cosmos-sdk/x/gov/types/v1"
 
 	"github.com/zeta-chain/node/e2e/txserver"
+	"github.com/zeta-chain/node/e2e/utils"
 )
 
 // ExecuteProposalSequence defines the sequence of proposals execution during the e2e tests
@@ -74,7 +75,15 @@ func (r *E2ERunner) CreateGovProposals(sequence ExecuteProposalSequence) error {
 		if err != nil {
 			return fmt.Errorf("failed to parse proposal file %s: %w", file.Name(), err)
 		}
-		r.Logger.Print("executing proposal: file name: %s title: %s", file.Name(), parsedProposal.Title)
+		r.Logger.Info("executing proposal: file name: %s title: %s", file.Name(), parsedProposal.Title)
+
+		minVersion := parsedProposal.MinVersion
+		zetacoredVersion := r.GetZetacoredVersion()
+
+		if !utils.MinimumVersionCheck(minVersion, zetacoredVersion) {
+			r.Logger.Print("⚠️ skipping proposal - %s (minimum version %s)", parsedProposal.Title, minVersion)
+			continue
+		}
 
 		// Create the proposal message
 		msg, err := govv1.NewMsgSubmitProposal(
@@ -151,12 +160,13 @@ func voteGovProposals(
 
 type proposal struct {
 	// Msgs defines an array of sdk.Msgs proto-JSON-encoded as Anys.
-	Messages  []json.RawMessage `json:"messages,omitempty"`
-	Metadata  string            `json:"metadata"`
-	Deposit   string            `json:"deposit"`
-	Title     string            `json:"title"`
-	Summary   string            `json:"summary"`
-	Expedited bool              `json:"expedited"`
+	Messages   []json.RawMessage `json:"messages,omitempty"`
+	Metadata   string            `json:"metadata"`
+	Deposit    string            `json:"deposit"`
+	Title      string            `json:"title"`
+	Summary    string            `json:"summary"`
+	Expedited  bool              `json:"expedited"`
+	MinVersion string            `json:"min_version,omitempty"`
 }
 
 // parseSubmitProposal reads and parses the proposal.

e2e/runner/run.go

@@ -4,16 +4,15 @@ import (
 	"fmt"
 	"time"
 
-	"golang.org/x/mod/semver"
+	"github.com/zeta-chain/node/e2e/utils"
 )
 
 // RunE2ETests runs a list of e2e tests
 func (r *E2ERunner) RunE2ETests(e2eTests []E2ETest) (err error) {
 	zetacoredVersion := r.GetZetacoredVersion()
 	for _, e2eTest := range e2eTests {
-		if semver.Major(zetacoredVersion) != "v0" && semver.Compare(zetacoredVersion, e2eTest.MinimumVersion) < 0 {
-			// note: spacing is padded to width of completed message
-			r.Logger.Print("⚠️ skipping  - %s (minimum version %s)", e2eTest.Name, e2eTest.MinimumVersion)
+		if !utils.MinimumVersionCheck(e2eTest.MinimumVersion, zetacoredVersion) {
+			r.Logger.Print("⚠️ skipping test - %s (minimum version %s)", e2eTest.Name, e2eTest.MinimumVersion)
 			continue
 		}
 		if err := r.Ctx.Err(); err != nil {

e2e/utils/utils.go

@@ -7,6 +7,7 @@ import (
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/mod/semver"
 )
 
 // ScriptPKToAddress is a hex string for P2WPKH script
@@ -45,3 +46,13 @@ func TestingFromContext(ctx context.Context) require.TestingT {
 
 	return t
 }
+
+func MinimumVersionCheck(testVersion, zetacoredVersion string) bool {
+	// If major version is "v0", return true regardless of comparison
+	if semver.Major(zetacoredVersion) == "v0" {
+		return true
+	}
+
+	// Otherwise, return true if zetacoredVersion >= testVersion
+	return semver.Compare(zetacoredVersion, testVersion) >= 0
+}

proto/zetachain/zetacore/emissions/params.proto

@@ -13,6 +13,7 @@ option go_package = "github.com/zeta-chain/node/x/emissions/types";
 //    ObserverSlashAmount:         100000000000000000,
 //    BallotMaturityBlocks:        100,
 //    BlockRewardAmount:           9620949074074074074.074070733466756687,
+//    PendingBallotsDeletionBufferBlocks: 144000
 message Params {
   option (gogoproto.goproto_stringer) = false;
   string validator_emission_percentage = 5;
@@ -27,6 +28,7 @@ message Params {
     (gogoproto.customtype) = "cosmossdk.io/math.LegacyDec",
     (gogoproto.nullable) = false
   ];
+  int64 pending_ballots_deletion_buffer_blocks = 12;
 
   // not used. do not edit.
   reserved 1 to 4;

testutil/keeper/mocks/emissions/observer.go

@@ -15,6 +15,7 @@ import { Message, proto3 } from "@bufbuild/protobuf";
  *    ObserverSlashAmount:         100000000000000000,
  *    BallotMaturityBlocks:        100,
  *    BlockRewardAmount:           9620949074074074074.074070733466756687,
+ *    PendingBallotsDeletionBufferBlocks: 144000
  *
  * @generated from message zetachain.zetacore.emissions.Params
  */
@@ -49,6 +50,11 @@ export declare class Params extends Message<Params> {
    */
   blockRewardAmount: string;
 
+  /**
+   * @generated from field: int64 pending_ballots_deletion_buffer_blocks = 12;
+   */
+  pendingBallotsDeletionBufferBlocks: bigint;
+
   constructor(data?: PartialMessage<Params>);
 
   static readonly runtime: typeof proto3;

typescript/zetachain/zetacore/emissions/params_pb.d.ts

@@ -107,10 +107,16 @@ func DistributeObserverRewards(
 	emissionsKeeper keeper.Keeper,
 	params types.Params,
 ) error {
+	// TODO : move the params BallotMaturityBlocks and BufferBlocksUnfinalizedBallots to the observer module
+	// https://github.com/zeta-chain/node/issues/3550
 	var (
-		slashAmount    = params.ObserverSlashAmount
+		slashAmount = params.ObserverSlashAmount
+		// Maturity blocks is used for distribution of rewards and deletion of finalized ballots
+		// and pending ballots at the maturity height, are simply ignored
 		maturityBlocks = params.BallotMaturityBlocks
-		maturedBallots []string
+		// The pendingBallotsDeletionBufferBlocks is a buffer number of blocks which is provided for pending ballots to allow them to be finalized
+		pendingBallotsDeletionBufferBlocks = params.PendingBallotsDeletionBufferBlocks
+		maturedBallots                     []string
 	)
 
 	err := emissionsKeeper.GetBankKeeper().
@@ -143,8 +149,14 @@ func DistributeObserverRewards(
 	// Processing Step 2: Emit the observer emissions
 	keeper.EmitObserverEmissions(ctx, finalDistributionList)
 
-	// Processing Step 3: Delete all matured ballots and the ballot list
-	emissionsKeeper.GetObserverKeeper().ClearMaturedBallotsAndBallotList(ctx, maturityBlocks)
+	// Processing Step 3a: Delete all finalized ballots at `maturityBlocksForFinalizedBallots` height
+	// This step optionally deletes the `BallotListForHeight` if all ballots are finalized and deleted
+	emissionsKeeper.GetObserverKeeper().ClearFinalizedMaturedBallots(ctx, maturityBlocks, false)
+
+	// Processing Step 3b: Delete all ballots at the buffered maturity height.
+	// This step deletes all remaining ballots and the `BallotListForHeight`.
+	bufferedMaturityBlocks := maturityBlocks + pendingBallotsDeletionBufferBlocks
+	emissionsKeeper.GetObserverKeeper().ClearFinalizedMaturedBallots(ctx, bufferedMaturityBlocks, true)
 	return nil
 }
 

x/emissions/abci.go

@@ -367,6 +367,29 @@ func TestDistributeObserverRewards(t *testing.T) {
 			slashAmount:     sdkmath.NewInt(25),
 			rewardsPerBlock: emissionstypes.BlockReward,
 		},
+		{
+			name: "no rewards if ballot is not finalized,irrespective of votes",
+			votes: [][]observertypes.VoteType{
+				{
+					observertypes.VoteType_NotYetVoted,
+					observertypes.VoteType_NotYetVoted,
+					observertypes.VoteType_SuccessObservation,
+					observertypes.VoteType_FailureObservation,
+				},
+			},
+			observerStartingEmissions: sdkmath.NewInt(100),
+			// total reward units would be 4 as all votes match the ballot status
+			totalRewardsForBlock: sdkmath.NewInt(0),
+			expectedRewards: map[string]int64{
+				observerSet.ObserverList[0]: 100,
+				observerSet.ObserverList[1]: 100,
+				observerSet.ObserverList[2]: 100,
+				observerSet.ObserverList[3]: 100,
+			},
+			ballotStatus:    observertypes.BallotStatus_BallotInProgress,
+			slashAmount:     sdkmath.NewInt(25),
+			rewardsPerBlock: emissionstypes.BlockReward,
+		},
 		{
 			name: "one observer slashed",
 			votes: [][]observertypes.VoteType{
@@ -537,6 +560,7 @@ func TestDistributeObserverRewards(t *testing.T) {
 	}
 	for _, tc := range tt {
 		t.Run(tc.name, func(t *testing.T) {
+			// Arrange
 			// Keeper initialization
 			k, ctx, sk, zk := keepertest.EmissionsKeeper(t)
 			zk.ObserverKeeper.SetObserverSet(ctx, observerSet)
@@ -581,10 +605,12 @@ func TestDistributeObserverRewards(t *testing.T) {
 			})
 			ctx = ctx.WithBlockHeight(100)
 
+			// Act
 			// Distribute the rewards and check if the rewards are distributed correctly
 			err = emissions.DistributeObserverRewards(ctx, tc.totalRewardsForBlock, *k, params)
-			require.NoError(t, err)
 
+			// Assert
+			require.NoError(t, err)
 			for i, observer := range observerSet.ObserverList {
 				observerEmission, found := k.GetWithdrawableEmission(ctx, observer)
 				require.True(t, found, "withdrawable emission not found for observer %d", i)
@@ -596,6 +622,12 @@ func TestDistributeObserverRewards(t *testing.T) {
 					i,
 				)
 			}
+			if tc.ballotStatus != observertypes.BallotStatus_BallotInProgress {
+				require.Len(t, zk.ObserverKeeper.GetAllBallots(ctx), 0)
+				_, found := zk.ObserverKeeper.GetBallotListForHeight(ctx, 0)
+				require.False(t, found)
+			}
+
 		})
 	}
 }

x/emissions/abci_test.go

@@ -4,8 +4,8 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 
 	"github.com/zeta-chain/node/x/emissions/exported"
-	v3 "github.com/zeta-chain/node/x/emissions/migrations/v3"
 	v4 "github.com/zeta-chain/node/x/emissions/migrations/v4"
+	v5 "github.com/zeta-chain/node/x/emissions/migrations/v5"
 )
 
 // Migrator is a struct for handling in-place store migrations.
@@ -22,12 +22,12 @@ func NewMigrator(k Keeper, ss exported.Subspace) Migrator {
 	}
 }
 
-// Migrate2to3 migrates the store from consensus version 2 to 3
-func (m Migrator) Migrate2to3(ctx sdk.Context) error {
-	return v3.MigrateStore(ctx, m.keeper, m.legacySubspace)
-}
-
 // Migrate3to4 migrates the store from consensus version 3 to 4
 func (m Migrator) Migrate3to4(ctx sdk.Context) error {
 	return v4.MigrateStore(ctx, m.keeper)
 }
+
+// Migrate4to5 migrates the store from consensus version 4 to 5
+func (m Migrator) Migrate4to5(ctx sdk.Context) error {
+	return v5.MigrateStore(ctx, m.keeper)
+}

x/emissions/keeper/migrator.go

@@ -40,7 +40,7 @@ func TestMsgServer_UpdateParams(t *testing.T) {
 		require.Error(t, err)
 	})
 
-	t.Run("fail for invalid params", func(t *testing.T) {
+	t.Run("fail for invalid params ,validatorEmissionPercentage is invalid", func(t *testing.T) {
 		k, ctx, _, _ := keepertest.EmissionsKeeper(t)
 		msgServer := keeper.NewMsgServerImpl(*k)
 		params := types.DefaultParams()
@@ -52,4 +52,17 @@ func TestMsgServer_UpdateParams(t *testing.T) {
 
 		require.ErrorIs(t, err, types.ErrUnableToSetParams)
 	})
+
+	t.Run("fail for invalid params ,pending buffer blocks is invalid", func(t *testing.T) {
+		k, ctx, _, _ := keepertest.EmissionsKeeper(t)
+		msgServer := keeper.NewMsgServerImpl(*k)
+		params := types.DefaultParams()
+		params.PendingBallotsDeletionBufferBlocks = -1
+		_, err := msgServer.UpdateParams(ctx, &types.MsgUpdateParams{
+			Authority: k.GetAuthority(),
+			Params:    params,
+		})
+
+		require.ErrorIs(t, err, types.ErrUnableToSetParams)
+	})
 }