feat: tally bundle proposal on skipUploaderRole

x/bundles/keeper/logic_bundles.go

@@ -3,6 +3,7 @@ package keeper
 import (
 	"cosmossdk.io/errors"
 	"cosmossdk.io/math"
+	poolTypes "github.com/KYVENetwork/chain/x/pool/types"
 
 	delegationTypes "github.com/KYVENetwork/chain/x/delegation/types"
 
@@ -510,3 +511,91 @@ func (k Keeper) GetVoteDistribution(ctx sdk.Context, poolId uint64) (voteDistrib
 
 	return
 }
+
+// tallyBundleProposal evaluates the votes of a bundle proposal and determines the outcome
+func (k msgServer) tallyBundleProposal(ctx sdk.Context, bundleProposal types.BundleProposal, poolId uint64) (types.TallyResult, error) {
+	// Increase points of stakers who did not vote at all + slash + remove if necessary.
+	// The protocol requires everybody to stay always active.
+	k.handleNonVoters(ctx, poolId)
+
+	// evaluate all votes and determine status based on the votes weighted with stake + delegation
+	voteDistribution := k.GetVoteDistribution(ctx, poolId)
+
+	// Handle tally outcome
+	switch voteDistribution.Status {
+	case types.BUNDLE_STATUS_VALID:
+		// charge the funders of the pool
+		fundersPayout, err := k.fundersKeeper.ChargeFundersOfPool(ctx, poolId)
+		if err != nil {
+			return types.TallyResult{}, err
+		}
+
+		// charge the inflation pool
+		inflationPayout, err := k.poolKeeper.ChargeInflationPool(ctx, poolId)
+		if err != nil {
+			return types.TallyResult{}, err
+		}
+
+		// calculate payouts to the different stakeholders like treasury, uploader and delegators
+		bundleReward := k.calculatePayouts(ctx, poolId, fundersPayout+inflationPayout)
+
+		// payout rewards to treasury
+		if err := util.TransferFromModuleToTreasury(k.accountKeeper, k.distrkeeper, ctx, poolTypes.ModuleName, bundleReward.Treasury); err != nil {
+			return types.TallyResult{}, err
+		}
+
+		// payout rewards to uploader through commission rewards
+		if err := k.stakerKeeper.IncreaseStakerCommissionRewards(ctx, bundleProposal.Uploader, bundleReward.Uploader); err != nil {
+			return types.TallyResult{}, err
+		}
+
+		// payout rewards to delegators through delegation rewards
+		if err := k.delegationKeeper.PayoutRewards(ctx, bundleProposal.Uploader, bundleReward.Delegation, poolTypes.ModuleName); err != nil {
+			return types.TallyResult{}, err
+		}
+
+		// slash stakers who voted incorrectly
+		for _, voter := range bundleProposal.VotersInvalid {
+			k.slashDelegatorsAndRemoveStaker(ctx, poolId, voter, delegationTypes.SLASH_TYPE_VOTE)
+		}
+
+		return types.TallyResult{
+			Status:           types.TallyResultValid,
+			VoteDistribution: voteDistribution,
+			FundersPayout:    fundersPayout,
+			InflationPayout:  inflationPayout,
+			BundleReward:     bundleReward,
+		}, nil
+	case types.BUNDLE_STATUS_INVALID:
+		// If the bundles is invalid, everybody who voted incorrectly gets slashed.
+		// The bundle provided by the message-sender is of no mean, because the previous bundle
+		// turned out to be incorrect.
+		// There this round needs to start again and the message-sender stays uploader.
+
+		// slash stakers who voted incorrectly - uploader receives upload slash
+		for _, voter := range bundleProposal.VotersValid {
+			if voter == bundleProposal.Uploader {
+				k.slashDelegatorsAndRemoveStaker(ctx, poolId, voter, delegationTypes.SLASH_TYPE_UPLOAD)
+			} else {
+				k.slashDelegatorsAndRemoveStaker(ctx, poolId, voter, delegationTypes.SLASH_TYPE_VOTE)
+			}
+		}
+
+		return types.TallyResult{
+			Status:           types.TallyResultInvalid,
+			VoteDistribution: voteDistribution,
+			FundersPayout:    0,
+			InflationPayout:  0,
+			BundleReward:     types.BundleReward{},
+		}, nil
+	default:
+		// If the bundle is neither valid nor invalid the quorum has not been reached yet.
+		return types.TallyResult{
+			Status:           types.TallyResultNoQuorum,
+			VoteDistribution: voteDistribution,
+			FundersPayout:    0,
+			InflationPayout:  0,
+			BundleReward:     types.BundleReward{},
+		}, nil
+	}
+}

x/bundles/keeper/msg_server_skip_uploader_role.go

@@ -15,19 +15,43 @@ func (k msgServer) SkipUploaderRole(goCtx context.Context, msg *types.MsgSkipUpl
 		return nil, err
 	}
 
-	pool, _ := k.poolKeeper.GetPool(ctx, msg.PoolId)
 	bundleProposal, _ := k.GetBundleProposal(ctx, msg.PoolId)
 
 	// reset points of uploader as node has proven to be active
 	k.resetPoints(ctx, msg.PoolId, msg.Staker)
 
+	// Previous round contains a bundle which needs to be validated now
+	result, err := k.tallyBundleProposal(ctx, bundleProposal, msg.PoolId)
+	if err != nil {
+		return nil, err
+	}
+
 	// Get next uploader, except the one who skipped
 	nextUploader := k.chooseNextUploader(ctx, msg.PoolId, msg.Staker)
 
-	bundleProposal.NextUploader = nextUploader
-	bundleProposal.UpdatedAt = uint64(ctx.BlockTime().Unix())
+	switch result.Status {
+	case types.TallyResultValid:
+		// Finalize bundle by adding it to the store
+		k.finalizeCurrentBundleProposal(ctx, msg.PoolId, result.VoteDistribution, result.FundersPayout, result.InflationPayout, result.BundleReward, nextUploader)
 
-	k.SetBundleProposal(ctx, bundleProposal)
+		// Register empty bundle with next uploader
+		bundleProposal = types.BundleProposal{
+			PoolId:       msg.PoolId,
+			NextUploader: nextUploader,
+			UpdatedAt:    uint64(ctx.BlockTime().Unix()),
+		}
+		k.SetBundleProposal(ctx, bundleProposal)
+	case types.TallyResultInvalid:
+		// Drop current bundle.
+		k.dropCurrentBundleProposal(ctx, msg.PoolId, result.VoteDistribution, nextUploader)
+	case types.TallyResultNoQuorum:
+		// Set next uploader and update the bundle proposal
+		bundleProposal.NextUploader = nextUploader
+		bundleProposal.UpdatedAt = uint64(ctx.BlockTime().Unix())
+		k.SetBundleProposal(ctx, bundleProposal)
+	}
+
+	pool, _ := k.poolKeeper.GetPool(ctx, msg.PoolId)
 
 	_ = ctx.EventManager().EmitTypedEvent(&types.EventSkippedUploaderRole{
 		PoolId:           msg.PoolId,

x/bundles/keeper/msg_server_skip_uploader_role_test.go

@@ -17,6 +17,8 @@ TEST CASES - msg_server_skip_uploader_role.go
 * Skip uploader role on data bundle if staker is next uploader
 * Skip uploader on data bundle after uploader role has already been skipped
 * Skip uploader role on dropped bundle
+* Skip uploader role on data bundle with current round containing a valid bundle
+* Skip uploader role on data bundle with current round containing an invalid bundle
 
 */
 
@@ -87,6 +89,17 @@ var _ = Describe("msg_server_skip_uploader_role.go", Ordered, func() {
 			PoolId:  0,
 		})
 
+		s.RunTxStakersSuccess(&stakertypes.MsgCreateStaker{
+			Creator: i.STAKER_2,
+			Amount:  100 * i.KYVE,
+		})
+
+		s.RunTxStakersSuccess(&stakertypes.MsgJoinPool{
+			Creator:    i.STAKER_2,
+			PoolId:     0,
+			Valaddress: i.VALADDRESS_2_A,
+		})
+
 		s.Commit()
 		s.WaitSeconds(60)
 
@@ -142,6 +155,10 @@ var _ = Describe("msg_server_skip_uploader_role.go", Ordered, func() {
 
 		// here the next uploader should be always be different after skipping
 		Expect(bundleProposal.NextUploader).To(Equal(i.STAKER_0))
+
+		// check that the bundle is not finalized
+		_, found = s.App().BundlesKeeper.GetFinalizedBundle(s.Ctx(), 0, 0)
+		Expect(found).To(BeFalse())
 	})
 
 	It("Skip uploader on data bundle after uploader role has already been skipped", func() {
@@ -186,7 +203,11 @@ var _ = Describe("msg_server_skip_uploader_role.go", Ordered, func() {
 		Expect(bundleProposal.VotersAbstain).To(BeEmpty())
 
 		// here the next uploader should be always be different after skipping
-		Expect(bundleProposal.NextUploader).To(Equal(i.STAKER_1))
+		Expect(bundleProposal.NextUploader).NotTo(Equal(i.STAKER_0))
+
+		// check that the bundle is not finalized
+		_, found = s.App().BundlesKeeper.GetFinalizedBundle(s.Ctx(), 0, 0)
+		Expect(found).To(BeFalse())
 	})
 
 	It("Skip uploader role on dropped bundle", func() {
@@ -225,6 +246,110 @@ var _ = Describe("msg_server_skip_uploader_role.go", Ordered, func() {
 		Expect(bundleProposal.VotersAbstain).To(BeEmpty())
 
 		// here the next uploader should be always be different after skipping
-		Expect(bundleProposal.NextUploader).To(Equal(i.STAKER_1))
+		Expect(bundleProposal.NextUploader).NotTo(Equal(i.STAKER_0))
+
+		// check that the bundle is not finalized
+		_, found = s.App().BundlesKeeper.GetFinalizedBundle(s.Ctx(), 0, 0)
+		Expect(found).To(BeFalse())
+	})
+
+	It("Skip uploader role on data bundle with current round containing a valid bundle", func() {
+		// ARRANGE
+		s.RunTxBundlesSuccess(&bundletypes.MsgVoteBundleProposal{
+			Creator:   i.VALADDRESS_1_A,
+			Staker:    i.STAKER_1,
+			PoolId:    0,
+			StorageId: "y62A3tfbSNcNYDGoL-eXwzyV-Zc9Q0OVtDvR1biJmNI",
+			Vote:      bundletypes.VOTE_TYPE_VALID,
+		})
+
+		s.Commit()
+		s.WaitSeconds(60)
+
+		// ACT
+		s.RunTxBundlesSuccess(&bundletypes.MsgSkipUploaderRole{
+			Creator:   i.VALADDRESS_1_A,
+			Staker:    i.STAKER_1,
+			PoolId:    0,
+			FromIndex: 100,
+		})
+
+		// ASSERT
+		bundleProposal, found := s.App().BundlesKeeper.GetBundleProposal(s.Ctx(), 0)
+		Expect(found).To(BeTrue())
+
+		Expect(bundleProposal.PoolId).To(Equal(uint64(0)))
+		Expect(bundleProposal.NextUploader).To(Equal(i.STAKER_0))
+		Expect(bundleProposal.UpdatedAt).NotTo(BeZero())
+		Expect(bundleProposal.StorageId).To(BeEmpty())
+		Expect(bundleProposal.Uploader).To(BeEmpty())
+		Expect(bundleProposal.DataSize).To(BeZero())
+		Expect(bundleProposal.DataHash).To(BeEmpty())
+		Expect(bundleProposal.BundleSize).To(BeZero())
+		Expect(bundleProposal.FromKey).To(BeEmpty())
+		Expect(bundleProposal.ToKey).To(BeEmpty())
+		Expect(bundleProposal.BundleSummary).To(BeEmpty())
+		Expect(bundleProposal.UpdatedAt).NotTo(BeZero())
+		Expect(bundleProposal.VotersValid).To(BeEmpty())
+		Expect(bundleProposal.VotersInvalid).To(BeEmpty())
+		Expect(bundleProposal.VotersAbstain).To(BeEmpty())
+
+		finalizedBundle, found := s.App().BundlesKeeper.GetFinalizedBundle(s.Ctx(), 0, 0)
+		Expect(found).To(BeTrue())
+
+		Expect(finalizedBundle.PoolId).To(Equal(uint64(0)))
+		Expect(finalizedBundle.Uploader).To(Equal(i.STAKER_0))
+	})
+
+	It("Skip uploader role on data bundle with current round containing an invalid bundle", func() {
+		// ARRANGE
+		s.RunTxBundlesSuccess(&bundletypes.MsgVoteBundleProposal{
+			Creator:   i.VALADDRESS_1_A,
+			Staker:    i.STAKER_1,
+			PoolId:    0,
+			StorageId: "y62A3tfbSNcNYDGoL-eXwzyV-Zc9Q0OVtDvR1biJmNI",
+			Vote:      bundletypes.VOTE_TYPE_INVALID,
+		})
+		s.RunTxBundlesSuccess(&bundletypes.MsgVoteBundleProposal{
+			Creator:   i.VALADDRESS_2_A,
+			Staker:    i.STAKER_2,
+			PoolId:    0,
+			StorageId: "y62A3tfbSNcNYDGoL-eXwzyV-Zc9Q0OVtDvR1biJmNI",
+			Vote:      bundletypes.VOTE_TYPE_INVALID,
+		})
+
+		s.Commit()
+		s.WaitSeconds(60)
+
+		// ACT
+		s.RunTxBundlesSuccess(&bundletypes.MsgSkipUploaderRole{
+			Creator:   i.VALADDRESS_1_A,
+			Staker:    i.STAKER_1,
+			PoolId:    0,
+			FromIndex: 100,
+		})
+
+		// ASSERT
+		bundleProposal, found := s.App().BundlesKeeper.GetBundleProposal(s.Ctx(), 0)
+		Expect(found).To(BeTrue())
+
+		Expect(bundleProposal.PoolId).To(Equal(uint64(0)))
+		Expect(bundleProposal.NextUploader).To(Equal(i.STAKER_2))
+		Expect(bundleProposal.StorageId).To(BeEmpty())
+		Expect(bundleProposal.Uploader).To(BeEmpty())
+		Expect(bundleProposal.DataSize).To(BeZero())
+		Expect(bundleProposal.DataHash).To(BeEmpty())
+		Expect(bundleProposal.BundleSize).To(BeZero())
+		Expect(bundleProposal.FromKey).To(BeEmpty())
+		Expect(bundleProposal.ToKey).To(BeEmpty())
+		Expect(bundleProposal.BundleSummary).To(BeEmpty())
+		Expect(bundleProposal.UpdatedAt).NotTo(BeZero())
+		Expect(bundleProposal.VotersValid).To(BeEmpty())
+		Expect(bundleProposal.VotersInvalid).To(BeEmpty())
+		Expect(bundleProposal.VotersAbstain).To(BeEmpty())
+
+		// check that the bundle is not finalized
+		_, found = s.App().BundlesKeeper.GetFinalizedBundle(s.Ctx(), 0, 0)
+		Expect(found).To(BeFalse())
 	})
 })