feat(x/protocolpool,x/distribution)!: remove dependency + fix continuous fund bug (#20790)

x/protocolpool/CHANGELOG.md

@@ -0,0 +1,30 @@
+<!--
+Guiding Principles:
+Changelogs are for humans, not machines.
+There should be an entry for every single version.
+The same types of changes should be grouped.
+Versions and sections should be linkable.
+The latest version comes first.
+The release date of each version is displayed.
+Mention whether you follow Semantic Versioning.
+Usage:
+Change log entries are to be added to the Unreleased section under the
+appropriate stanza (see below). Each entry should ideally include a tag and
+the Github issue reference in the following format:
+* (<tag>) [#<issue-number>] Changelog message.
+Types of changes (Stanzas):
+"Features" for new features.
+"Improvements" for changes in existing functionality.
+"Deprecated" for soon-to-be removed features.
+"Bug Fixes" for any bug fixes.
+"API Breaking" for breaking exported APIs used by developers building on SDK.
+Ref: https://keepachangelog.com/en/1.0.0/
+-->
+
+# Changelog
+
+## [Unreleased]
+
+### Improvements
+
+* [#20790](https://github.com/cosmos/cosmos-sdk/pull/20790) `x/protocolpool` now has its own BeginBlock.

x/protocolpool/keeper/genesis.go

@@ -3,7 +3,9 @@ package keeper
 import (
 	"context"
 	"fmt"
+	"time"
 
+	"cosmossdk.io/math"
 	"cosmossdk.io/x/protocolpool/types"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -49,8 +51,21 @@ func (k Keeper) InitGenesis(ctx context.Context, data *types.GenesisState) error
 		}
 	}
 
-	if err := k.ToDistribute.Set(ctx, data.ToDistribute); err != nil {
-		return fmt.Errorf("failed to set to distribute: %w", err)
+	if err := k.LastBalance.Set(ctx, data.LastBalance); err != nil {
+		return fmt.Errorf("failed to set last balance: %w", err)
+	}
+
+	totalToBeDistributed := math.ZeroInt()
+	for _, distribution := range data.Distributions {
+		totalToBeDistributed = totalToBeDistributed.Add(distribution.Amount)
+		if err := k.Distributions.Set(ctx, *distribution.Time, distribution.Amount); err != nil {
+			return fmt.Errorf("failed to set distribution: %w", err)
+		}
+	}
+
+	// sanity check to avoid trying to distribute more than what is available
+	if data.LastBalance.LT(totalToBeDistributed) {
+		return fmt.Errorf("total to be distributed is greater than the last balance")
 	}
 
 	return nil
@@ -96,7 +111,19 @@ func (k Keeper) ExportGenesis(ctx context.Context) (*types.GenesisState, error)
 
 	genState := types.NewGenesisState(cf, budget)
 
-	genState.ToDistribute, err = k.ToDistribute.Get(ctx)
+	genState.LastBalance, err = k.LastBalance.Get(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	err = k.Distributions.Walk(ctx, nil, func(key time.Time, value math.Int) (stop bool, err error) {
+		genState.Distributions = append(genState.Distributions, &types.Distribution{
+			Time:   &key,
+			Amount: value,
+		})
+
+		return false, nil
+	})
 	if err != nil {
 		return nil, err
 	}

x/protocolpool/keeper/genesis_test.go

@@ -31,7 +31,17 @@ func (suite *KeeperTestSuite) TestInitGenesis() {
 		},
 	)
 
+	gs.Distributions = append(gs.Distributions, &types.Distribution{
+		Amount: math.OneInt(),
+		Time:   &time.Time{},
+	})
+
 	err := suite.poolKeeper.InitGenesis(suite.ctx, gs)
+	suite.Require().ErrorContains(err, "total to be distributed is greater than the last balance")
+
+	// Set last balance
+	gs.LastBalance = math.NewInt(1)
+	err = suite.poolKeeper.InitGenesis(suite.ctx, gs)
 	suite.Require().NoError(err)
 
 	// Export
@@ -39,4 +49,5 @@ func (suite *KeeperTestSuite) TestInitGenesis() {
 	suite.Require().NoError(err)
 	suite.Require().Equal(gs.ContinuousFund, exportedGenState.ContinuousFund)
 	suite.Require().Equal(gs.Budget, exportedGenState.Budget)
+	suite.Require().Equal(math.OneInt(), exportedGenState.LastBalance)
 }

x/protocolpool/keeper/keeper.go

@@ -1,10 +1,10 @@
 package keeper
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"sort"
 	"time"
 
 	"cosmossdk.io/collections"
@@ -35,8 +35,8 @@ type Keeper struct {
 	ContinuousFund collections.Map[sdk.AccAddress, types.ContinuousFund]
 	// RecipientFundDistribution key: RecipientAddr | value: Claimable amount
 	RecipientFundDistribution collections.Map[sdk.AccAddress, math.Int]
-	// ToDistribute is to keep track of funds to be distributed. It gets zeroed out in iterateAndUpdateFundsDistribution.
-	ToDistribute collections.Item[math.Int]
+	Distributions             collections.Map[time.Time, math.Int] // key: time.Time | value: amount
+	LastBalance               collections.Item[math.Int]
 }
 
 func NewKeeper(cdc codec.BinaryCodec, env appmodule.Environment, ak types.AccountKeeper, bk types.BankKeeper, sk types.StakingKeeper, authority string,
@@ -49,6 +49,10 @@ func NewKeeper(cdc codec.BinaryCodec, env appmodule.Environment, ak types.Accoun
 	if addr := ak.GetModuleAddress(types.StreamAccount); addr == nil {
 		panic(fmt.Sprintf("%s module account has not been set", types.StreamAccount))
 	}
+	// ensure protocol pool distribution account is set
+	if addr := ak.GetModuleAddress(types.ProtocolPoolDistrAccount); addr == nil {
+		panic(fmt.Sprintf("%s module account has not been set", types.ProtocolPoolDistrAccount))
+	}
 
 	sb := collections.NewSchemaBuilder(env.KVStoreService)
 
@@ -62,7 +66,8 @@ func NewKeeper(cdc codec.BinaryCodec, env appmodule.Environment, ak types.Accoun
 		BudgetProposal:            collections.NewMap(sb, types.BudgetKey, "budget", sdk.AccAddressKey, codec.CollValue[types.Budget](cdc)),
 		ContinuousFund:            collections.NewMap(sb, types.ContinuousFundKey, "continuous_fund", sdk.AccAddressKey, codec.CollValue[types.ContinuousFund](cdc)),
 		RecipientFundDistribution: collections.NewMap(sb, types.RecipientFundDistributionKey, "recipient_fund_distribution", sdk.AccAddressKey, sdk.IntValue),
-		ToDistribute:              collections.NewItem(sb, types.ToDistributeKey, "to_distribute", sdk.IntValue),
+		Distributions:             collections.NewMap(sb, types.DistributionsKey, "distributions", sdk.TimeKey, sdk.IntValue),
+		LastBalance:               collections.NewItem(sb, types.LastBalanceKey, "last_balance", sdk.IntValue),
 	}
 
 	schema, err := sb.Build()
@@ -80,19 +85,19 @@ func (k Keeper) GetAuthority() string {
 }
 
 // FundCommunityPool allows an account to directly fund the community fund pool.
-func (k Keeper) FundCommunityPool(ctx context.Context, amount sdk.Coins, sender sdk.AccAddress) error {
+func (k Keeper) FundCommunityPool(ctx context.Context, amount sdk.Coins, sender []byte) error {
 	return k.bankKeeper.SendCoinsFromAccountToModule(ctx, sender, types.ModuleName, amount)
 }
 
 // DistributeFromCommunityPool distributes funds from the protocolpool module account to
 // a receiver address.
-func (k Keeper) DistributeFromCommunityPool(ctx context.Context, amount sdk.Coins, receiveAddr sdk.AccAddress) error {
+func (k Keeper) DistributeFromCommunityPool(ctx context.Context, amount sdk.Coins, receiveAddr []byte) error {
 	return k.bankKeeper.SendCoinsFromModuleToAccount(ctx, types.ModuleName, receiveAddr, amount)
 }
 
 // DistributeFromStreamFunds distributes funds from the protocolpool's stream module account to
 // a receiver address.
-func (k Keeper) DistributeFromStreamFunds(ctx context.Context, amount sdk.Coins, receiveAddr sdk.AccAddress) error {
+func (k Keeper) DistributeFromStreamFunds(ctx context.Context, amount sdk.Coins, receiveAddr []byte) error {
 	return k.bankKeeper.SendCoinsFromModuleToAccount(ctx, types.StreamAccount, receiveAddr, amount)
 }
 
@@ -100,48 +105,17 @@ func (k Keeper) DistributeFromStreamFunds(ctx context.Context, amount sdk.Coins,
 func (k Keeper) GetCommunityPool(ctx context.Context) (sdk.Coins, error) {
 	moduleAccount := k.authKeeper.GetModuleAccount(ctx, types.ModuleName)
 	if moduleAccount == nil {
-		return nil, errorsmod.Wrapf(sdkerrors.ErrUnknownAddress, "module account %s does not exist", moduleAccount)
+		return nil, errorsmod.Wrapf(sdkerrors.ErrUnknownAddress, "module account %s does not exist", types.ModuleName)
 	}
 	return k.bankKeeper.GetAllBalances(ctx, moduleAccount.GetAddress()), nil
 }
 
-func (k Keeper) withdrawContinuousFund(ctx context.Context, recipientAddr string) (sdk.Coin, error) {
-	recipient, err := k.authKeeper.AddressCodec().StringToBytes(recipientAddr)
-	if err != nil {
-		return sdk.Coin{}, sdkerrors.ErrInvalidAddress.Wrapf("invalid recipient address: %s", err)
-	}
-
-	cf, err := k.ContinuousFund.Get(ctx, recipient)
-	if err != nil {
-		if errors.Is(err, collections.ErrNotFound) {
-			return sdk.Coin{}, fmt.Errorf("no continuous fund found for recipient: %s", recipientAddr)
-		}
-		return sdk.Coin{}, fmt.Errorf("get continuous fund failed for recipient: %s", recipientAddr)
-	}
-	if cf.Expiry != nil && cf.Expiry.Before(k.HeaderService.HeaderInfo(ctx).Time) {
-		return sdk.Coin{}, fmt.Errorf("cannot withdraw continuous funds: continuous fund expired for recipient: %s", recipientAddr)
-	}
-
-	err = k.IterateAndUpdateFundsDistribution(ctx)
-	if err != nil {
-		return sdk.Coin{}, fmt.Errorf("error while iterating all the continuous funds: %w", err)
-	}
-
-	// withdraw continuous fund
-	withdrawnAmount, err := k.withdrawRecipientFunds(ctx, recipient)
-	if err != nil {
-		return sdk.Coin{}, fmt.Errorf("error while withdrawing recipient funds for recipient: %s", recipientAddr)
-	}
-
-	return withdrawnAmount, nil
-}
-
 func (k Keeper) withdrawRecipientFunds(ctx context.Context, recipient []byte) (sdk.Coin, error) {
 	// get allocated continuous fund
 	fundsAllocated, err := k.RecipientFundDistribution.Get(ctx, recipient)
 	if err != nil {
 		if errors.Is(err, collections.ErrNotFound) {
-			return sdk.Coin{}, types.ErrNoRecipientFund
+			return sdk.Coin{}, types.ErrNoRecipientFound
 		}
 		return sdk.Coin{}, err
 	}
@@ -166,20 +140,12 @@ func (k Keeper) withdrawRecipientFunds(ctx context.Context, recipient []byte) (s
 	return withdrawnAmount, nil
 }
 
-// SetToDistribute sets the amount to be distributed among recipients, usually called by x/distribution while allocating
-// reward and fee distribution.
-// This could be only set by the authority address.
-func (k Keeper) SetToDistribute(ctx context.Context, amount sdk.Coins, addr string) error {
-	authAddr, err := k.authKeeper.AddressCodec().StringToBytes(addr)
-	if err != nil {
-		return err
-	}
-	hasPermission, err := k.hasPermission(authAddr)
-	if err != nil {
-		return err
-	}
-	if !hasPermission {
-		return sdkerrors.ErrUnauthorized
+// SetToDistribute sets the amount to be distributed among recipients.
+func (k Keeper) SetToDistribute(ctx context.Context) error {
+	// Get current balance of the intermediary module account
+	moduleAccount := k.authKeeper.GetModuleAccount(ctx, types.ProtocolPoolDistrAccount)
+	if moduleAccount == nil {
+		return errorsmod.Wrapf(sdkerrors.ErrUnknownAddress, "module account %s does not exist", types.ProtocolPoolDistrAccount)
 	}
 
 	denom, err := k.stakingKeeper.BondDenom(ctx)
@@ -187,106 +153,44 @@ func (k Keeper) SetToDistribute(ctx context.Context, amount sdk.Coins, addr stri
 		return err
 	}
 
-	totalStreamFundsPercentage := math.LegacyZeroDec()
-	err = k.ContinuousFund.Walk(ctx, nil, func(key sdk.AccAddress, cf types.ContinuousFund) (stop bool, err error) {
-		// Check if the continuous fund has expired
-		if cf.Expiry != nil && cf.Expiry.Before(k.HeaderService.HeaderInfo(ctx).Time) {
-			return false, nil
-		}
+	currentBalance := k.bankKeeper.GetAllBalances(ctx, moduleAccount.GetAddress())
+	distributionBalance := currentBalance.AmountOf(denom)
 
-		totalStreamFundsPercentage = totalStreamFundsPercentage.Add(cf.Percentage)
-		if totalStreamFundsPercentage.GT(math.LegacyOneDec()) {
-			return true, errors.New("total funds percentage cannot exceed 100")
-		}
-
-		return false, nil
-	})
-	if err != nil {
-		return err
-	}
-
-	// if percentage is 0 then return early
-	if totalStreamFundsPercentage.IsZero() {
+	// if the balance is zero, return early
+	if distributionBalance.IsZero() {
 		return nil
 	}
 
-	// send streaming funds to the stream module account
-	toDistributeAmt := math.LegacyNewDecFromInt(amount.AmountOf(denom)).Mul(totalStreamFundsPercentage).TruncateInt()
-	streamAmt := sdk.NewCoins(sdk.NewCoin(denom, toDistributeAmt))
-	if err := k.bankKeeper.SendCoinsFromModuleToModule(ctx, types.ModuleName, types.StreamAccount, streamAmt); err != nil {
-		return err
-	}
-
-	amountToDistribute, err := k.ToDistribute.Get(ctx)
+	lastBalance, err := k.LastBalance.Get(ctx)
 	if err != nil {
 		if errors.Is(err, collections.ErrNotFound) {
-			amountToDistribute = math.ZeroInt()
+			lastBalance = math.ZeroInt()
 		} else {
 			return err
 		}
 	}
 
-	err = k.ToDistribute.Set(ctx, amountToDistribute.Add(amount.AmountOf(denom)))
-	if err != nil {
-		return fmt.Errorf("error while setting ToDistribute: %w", err)
-	}
-	return nil
-}
+	// Calculate the amount to be distributed
+	amountToDistribute := distributionBalance.Sub(lastBalance)
 
-func (k Keeper) hasPermission(addr []byte) (bool, error) {
-	authority := k.GetAuthority()
-	authAcc, err := k.authKeeper.AddressCodec().StringToBytes(authority)
-	if err != nil {
-		return false, err
+	if err = k.Distributions.Set(ctx, k.HeaderService.HeaderInfo(ctx).Time, amountToDistribute); err != nil {
+		return fmt.Errorf("error while setting Distributions: %w", err)
 	}
 
-	return bytes.Equal(authAcc, addr), nil
+	// Update the last balance
+	return k.LastBalance.Set(ctx, distributionBalance)
 }
 
 func (k Keeper) IterateAndUpdateFundsDistribution(ctx context.Context) error {
-	toDistributeAmount, err := k.ToDistribute.Get(ctx)
-	if err != nil {
-		return err
-	}
+	// first we get all the continuous funds, and keep a list of the ones that expired so we can delete later
+	funds := []types.ContinuousFund{}
+	toDelete := [][]byte{}
+	err := k.ContinuousFund.Walk(ctx, nil, func(key sdk.AccAddress, cf types.ContinuousFund) (stop bool, err error) {
+		funds = append(funds, cf)
 
-	// if there are no funds to distribute, return
-	if toDistributeAmount.IsZero() {
-		return nil
-	}
-
-	totalPercentageToBeDistributed := math.LegacyZeroDec()
-
-	denom, err := k.stakingKeeper.BondDenom(ctx)
-	if err != nil {
-		return err
-	}
-	toDistributeDec := sdk.NewDecCoin(denom, toDistributeAmount)
-
-	// Calculate totalPercentageToBeDistributed and store values
-	err = k.ContinuousFund.Walk(ctx, nil, func(key sdk.AccAddress, cf types.ContinuousFund) (stop bool, err error) {
-		// Check if the continuous fund has expired
+		// check if the continuous fund has expired, and add it to the list of funds to delete
 		if cf.Expiry != nil && cf.Expiry.Before(k.HeaderService.HeaderInfo(ctx).Time) {
-			return false, nil
-		}
-
-		// sanity check for max percentage
-		totalPercentageToBeDistributed = totalPercentageToBeDistributed.Add(cf.Percentage)
-		if totalPercentageToBeDistributed.GT(math.LegacyOneDec()) {
-			return true, errors.New("total funds percentage cannot exceed 100")
-		}
-
-		// Calculate the funds to be distributed based on the percentage
-		recipientAmount := toDistributeDec.Amount.Mul(cf.Percentage).TruncateInt()
-
-		// Set funds to be claimed
-		toClaim, err := k.RecipientFundDistribution.Get(ctx, key)
-		if err != nil {
-			return true, err
-		}
-		amount := toClaim.Add(recipientAmount)
-		err = k.RecipientFundDistribution.Set(ctx, key, amount)
-		if err != nil {
-			return true, err
+			toDelete = append(toDelete, key)
 		}
 
 		return false, nil
@@ -295,8 +199,108 @@ func (k Keeper) IterateAndUpdateFundsDistribution(ctx context.Context) error {
 		return err
 	}
 
-	// Set the coins to be distributed from toDistribute to 0
-	return k.ToDistribute.Set(ctx, math.ZeroInt())
+	// next we iterate over the distributions, calculate each recipient's share and the remaining pool funds
+	toDistribute := map[string]math.Int{}
+	poolFunds := math.ZeroInt()
+	fullAmountToDistribute := math.ZeroInt()
+
+	if err = k.Distributions.Walk(ctx, nil, func(key time.Time, amount math.Int) (stop bool, err error) {
+		percentageToDistribute := math.LegacyZeroDec()
+		for _, f := range funds {
+			if f.Expiry != nil && f.Expiry.Before(key) {
+				continue
+			}
+
+			percentageToDistribute = percentageToDistribute.Add(f.Percentage)
+
+			_, ok := toDistribute[f.Recipient]
+			if !ok {
+				toDistribute[f.Recipient] = math.ZeroInt()
+			}
+			amountToDistribute := f.Percentage.MulInt(amount).TruncateInt()
+			toDistribute[f.Recipient] = toDistribute[f.Recipient].Add(amountToDistribute)
+			fullAmountToDistribute = fullAmountToDistribute.Add(amountToDistribute)
+		}
+
+		// sanity check for max percentage
+		if percentageToDistribute.GT(math.LegacyOneDec()) {
+			return true, errors.New("total funds percentage cannot exceed 100")
+		}
+
+		remaining := math.LegacyOneDec().Sub(percentageToDistribute).MulInt(amount).RoundInt()
+		poolFunds = poolFunds.Add(remaining)
+
+		return false, nil
+	}); err != nil {
+		return err
+	}
+
+	// clear the distributions and reset the last balance
+	if err = k.Distributions.Clear(ctx, nil); err != nil {
+		return err
+	}
+
+	if err = k.LastBalance.Set(ctx, math.ZeroInt()); err != nil {
+		return err
+	}
+
+	// send the funds to the stream account to be distributed later, and the remaining to the community pool
+	bondDenom, err := k.stakingKeeper.BondDenom(ctx)
+	if err != nil {
+		return err
+	}
+
+	streamAmt := sdk.NewCoins(sdk.NewCoin(bondDenom, fullAmountToDistribute))
+	if !streamAmt.IsZero() {
+		if err := k.bankKeeper.SendCoinsFromModuleToModule(ctx, types.ProtocolPoolDistrAccount, types.StreamAccount, streamAmt); err != nil {
+			return err
+		}
+	}
+
+	if !poolFunds.IsZero() {
+		poolCoins := sdk.NewCoins(sdk.NewCoin(bondDenom, poolFunds))
+		if err := k.bankKeeper.SendCoinsFromModuleToModule(ctx, types.ProtocolPoolDistrAccount, types.ModuleName, poolCoins); err != nil {
+			return err
+		}
+	}
+
+	// update the recipient fund distribution, first get the keys and sort them
+	recipients := make([]string, 0, len(toDistribute))
+	for k2 := range toDistribute {
+		recipients = append(recipients, k2)
+	}
+	sort.Strings(recipients)
+
+	for _, recipient := range recipients {
+		// Set funds to be claimed
+		bzAddr, err := k.authKeeper.AddressCodec().StringToBytes(recipient)
+		if err != nil {
+			return err
+		}
+
+		toClaim, err := k.RecipientFundDistribution.Get(ctx, bzAddr)
+		if err != nil {
+			if errors.Is(err, collections.ErrNotFound) {
+				toClaim = math.ZeroInt()
+			} else {
+				return err
+			}
+		}
+
+		amount := toClaim.Add(toDistribute[recipient])
+		if err = k.RecipientFundDistribution.Set(ctx, bzAddr, amount); err != nil {
+			return err
+		}
+	}
+
+	// delete expired continuous funds
+	for _, recipient := range toDelete {
+		if err = k.ContinuousFund.Remove(ctx, recipient); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func (k Keeper) claimFunds(ctx context.Context, recipientAddr string) (amount sdk.Coin, err error) {
@@ -472,3 +476,7 @@ func (k Keeper) validateContinuousFund(ctx context.Context, msg types.MsgCreateC
 
 	return nil
 }
+
+func (k Keeper) BeginBlocker(ctx context.Context) error {
+	return k.SetToDistribute(ctx)
+}

x/protocolpool/keeper/keeper_test.go

@@ -27,8 +27,9 @@ import (
 )
 
 var (
-	poolAcc   = authtypes.NewEmptyModuleAccount(types.ModuleName)
-	streamAcc = authtypes.NewEmptyModuleAccount(types.StreamAccount)
+	poolAcc      = authtypes.NewEmptyModuleAccount(types.ModuleName)
+	streamAcc    = authtypes.NewEmptyModuleAccount(types.StreamAccount)
+	poolDistrAcc = authtypes.NewEmptyModuleAccount(types.ProtocolPoolDistrAccount)
 )
 
 type KeeperTestSuite struct {
@@ -57,6 +58,7 @@ func (s *KeeperTestSuite) SetupTest() {
 	ctrl := gomock.NewController(s.T())
 	accountKeeper := pooltestutil.NewMockAccountKeeper(ctrl)
 	accountKeeper.EXPECT().GetModuleAddress(types.ModuleName).Return(poolAcc.GetAddress())
+	accountKeeper.EXPECT().GetModuleAddress(types.ProtocolPoolDistrAccount).Return(poolDistrAcc.GetAddress())
 	accountKeeper.EXPECT().AddressCodec().Return(address.NewBech32Codec("cosmos")).AnyTimes()
 	accountKeeper.EXPECT().GetModuleAddress(types.StreamAccount).Return(streamAcc.GetAddress())
 	s.authKeeper = accountKeeper
@@ -102,12 +104,14 @@ func (s *KeeperTestSuite) mockWithdrawContinuousFund() {
 	s.stakingKeeper.EXPECT().BondDenom(gomock.Any()).Return("stake", nil).AnyTimes()
 }
 
-func (s *KeeperTestSuite) mockStreamFunds() {
+func (s *KeeperTestSuite) mockStreamFunds(distributed math.Int) {
 	s.authKeeper.EXPECT().GetModuleAccount(s.ctx, types.ModuleName).Return(poolAcc).AnyTimes()
+	s.authKeeper.EXPECT().GetModuleAccount(s.ctx, types.ProtocolPoolDistrAccount).Return(poolDistrAcc).AnyTimes()
 	s.authKeeper.EXPECT().GetModuleAddress(types.StreamAccount).Return(streamAcc.GetAddress()).AnyTimes()
-	distrBal := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(100000)))
-	s.bankKeeper.EXPECT().GetAllBalances(s.ctx, poolAcc.GetAddress()).Return(distrBal).AnyTimes()
-	s.bankKeeper.EXPECT().SendCoinsFromModuleToModule(s.ctx, poolAcc.GetName(), streamAcc.GetName(), gomock.Any()).AnyTimes()
+	distrBal := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, distributed))
+	s.bankKeeper.EXPECT().GetAllBalances(s.ctx, poolDistrAcc.GetAddress()).Return(distrBal).AnyTimes()
+	s.bankKeeper.EXPECT().SendCoinsFromModuleToModule(s.ctx, poolDistrAcc.GetName(), streamAcc.GetName(), gomock.Any()).AnyTimes()
+	s.bankKeeper.EXPECT().SendCoinsFromModuleToModule(s.ctx, poolDistrAcc.GetName(), poolAcc.GetName(), gomock.Any()).AnyTimes()
 }
 
 func TestKeeperTestSuite(t *testing.T) {
@@ -118,10 +122,11 @@ func (s *KeeperTestSuite) TestIterateAndUpdateFundsDistribution() {
 	// We'll create 2 continuous funds of 30% each, and the total pool is 1000000, meaning each fund should get 300000
 
 	s.SetupTest()
-	s.authKeeper.EXPECT().GetModuleAccount(s.ctx, types.ModuleName).Return(poolAcc).AnyTimes()
+	s.authKeeper.EXPECT().GetModuleAccount(s.ctx, types.ProtocolPoolDistrAccount).Return(poolAcc).AnyTimes()
 	distrBal := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(1000000)))
 	s.bankKeeper.EXPECT().GetAllBalances(s.ctx, poolAcc.GetAddress()).Return(distrBal).AnyTimes()
-	s.bankKeeper.EXPECT().SendCoinsFromModuleToModule(s.ctx, poolAcc.GetName(), streamAcc.GetName(), sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(600000)))).AnyTimes()
+	s.bankKeeper.EXPECT().SendCoinsFromModuleToModule(s.ctx, poolDistrAcc.GetName(), streamAcc.GetName(), sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(600000))))
+	s.bankKeeper.EXPECT().SendCoinsFromModuleToModule(s.ctx, poolDistrAcc.GetName(), poolAcc.GetName(), sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(400000))))
 
 	_, err := s.msgServer.CreateContinuousFund(s.ctx, &types.MsgCreateContinuousFund{
 		Authority:  s.poolKeeper.GetAuthority(),
@@ -137,7 +142,7 @@ func (s *KeeperTestSuite) TestIterateAndUpdateFundsDistribution() {
 	})
 	s.Require().NoError(err)
 
-	_ = s.poolKeeper.SetToDistribute(s.ctx, sdk.NewCoins(sdk.NewCoin("stake", math.NewInt(1000000))), s.poolKeeper.GetAuthority())
+	_ = s.poolKeeper.SetToDistribute(s.ctx)
 
 	err = s.poolKeeper.IterateAndUpdateFundsDistribution(s.ctx)
 	s.Require().NoError(err)

x/protocolpool/keeper/msg_server.go

@@ -139,6 +139,11 @@ func (k MsgServer) CreateContinuousFund(ctx context.Context, msg *types.MsgCreat
 		return nil, fmt.Errorf("cannot set continuous fund proposal\ntotal funds percentage exceeds 100\ncurrent total percentage: %s", totalStreamFundsPercentage.Sub(msg.Percentage).MulInt64(100).TruncateInt().String())
 	}
 
+	// Distribute funds to avoid giving this new fund more than it should get
+	if err := k.IterateAndUpdateFundsDistribution(ctx); err != nil {
+		return nil, err
+	}
+
 	// Create continuous fund proposal
 	cf := types.ContinuousFund{
 		Recipient:  msg.Recipient,
@@ -161,12 +166,23 @@ func (k MsgServer) CreateContinuousFund(ctx context.Context, msg *types.MsgCreat
 }
 
 func (k MsgServer) WithdrawContinuousFund(ctx context.Context, msg *types.MsgWithdrawContinuousFund) (*types.MsgWithdrawContinuousFundResponse, error) {
-	amount, err := k.withdrawContinuousFund(ctx, msg.RecipientAddress)
+	recipient, err := k.authKeeper.AddressCodec().StringToBytes(msg.RecipientAddress)
 	if err != nil {
-		return nil, err
+		return nil, sdkerrors.ErrInvalidAddress.Wrapf("invalid recipient address: %s", err)
 	}
 
-	return &types.MsgWithdrawContinuousFundResponse{Amount: amount}, nil
+	err = k.IterateAndUpdateFundsDistribution(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("error while iterating all the continuous funds: %w", err)
+	}
+
+	// withdraw continuous fund
+	withdrawnAmount, err := k.withdrawRecipientFunds(ctx, recipient)
+	if err != nil {
+		return nil, fmt.Errorf("error while withdrawing recipient funds for recipient: %w", err)
+	}
+
+	return &types.MsgWithdrawContinuousFundResponse{Amount: withdrawnAmount}, nil
 }
 
 func (k MsgServer) CancelContinuousFund(ctx context.Context, msg *types.MsgCancelContinuousFund) (*types.MsgCancelContinuousFundResponse, error) {
@@ -182,17 +198,14 @@ func (k MsgServer) CancelContinuousFund(ctx context.Context, msg *types.MsgCance
 	canceledHeight := k.HeaderService.HeaderInfo(ctx).Height
 	canceledTime := k.HeaderService.HeaderInfo(ctx).Time
 
-	found, err := k.ContinuousFund.Has(ctx, recipient)
-	if !found {
-		return nil, fmt.Errorf("no recipient found to cancel continuous fund: %s", msg.RecipientAddress)
-	}
-	if err != nil {
+	// distribute funds before withdrawing
+	if err = k.IterateAndUpdateFundsDistribution(ctx); err != nil {
 		return nil, err
 	}
 
 	// withdraw funds if any are allocated
 	withdrawnFunds, err := k.withdrawRecipientFunds(ctx, recipient)
-	if err != nil && !errorspkg.Is(err, types.ErrNoRecipientFund) {
+	if err != nil && !errorspkg.Is(err, types.ErrNoRecipientFound) {
 		return nil, fmt.Errorf("error while withdrawing already allocated funds for recipient %s: %w", msg.RecipientAddress, err)
 	}
 

x/protocolpool/keeper/msg_server_test.go

@@ -3,6 +3,8 @@ package keeper_test
 import (
 	"time"
 
+	"github.com/golang/mock/gomock"
+
 	"cosmossdk.io/collections"
 	"cosmossdk.io/core/header"
 	"cosmossdk.io/math"
@@ -403,7 +405,7 @@ func (suite *KeeperTestSuite) TestWithdrawContinuousFund() {
 		"recipient with no continuous fund": {
 			recipientAddress: []sdk.AccAddress{recipient},
 			expErr:           true,
-			expErrMsg:        "no continuous fund found for recipient",
+			expErrMsg:        "error while withdrawing recipient funds for recipient: no recipient found",
 		},
 		"funds percentage > 100": {
 			preRun: func() {
@@ -441,14 +443,14 @@ func (suite *KeeperTestSuite) TestWithdrawContinuousFund() {
 				suite.Require().NoError(err)
 
 				// Set ToDistribute
-				err = suite.poolKeeper.ToDistribute.Set(suite.ctx, math.NewInt(100000))
+				err = suite.poolKeeper.Distributions.Set(suite.ctx, suite.ctx.HeaderInfo().Time, math.NewInt(100000))
 				suite.Require().NoError(err)
 			},
 			recipientAddress: []sdk.AccAddress{recipient},
 			expErr:           true,
 			expErrMsg:        "error while iterating all the continuous funds: total funds percentage cannot exceed 100",
 		},
-		"expired case": {
+		"expired case with no funds left to withdraw": {
 			preRun: func() {
 				percentage, err := math.LegacyNewDecFromStr("0.2")
 				suite.Require().NoError(err)
@@ -464,7 +466,7 @@ func (suite *KeeperTestSuite) TestWithdrawContinuousFund() {
 			},
 			recipientAddress: []sdk.AccAddress{recipient},
 			expErr:           true,
-			expErrMsg:        "cannot withdraw continuous funds: continuous fund expired for recipient",
+			expErrMsg:        "error while withdrawing recipient funds for recipient: no recipient found",
 		},
 		"valid case with ToDistribute amount zero": {
 			preRun: func() {
@@ -483,8 +485,9 @@ func (suite *KeeperTestSuite) TestWithdrawContinuousFund() {
 				// Set recipient fund percentage and recipient fund distribution
 				err = suite.poolKeeper.RecipientFundDistribution.Set(suite.ctx, recipient, math.ZeroInt())
 				suite.Require().NoError(err)
-				err = suite.poolKeeper.ToDistribute.Set(suite.ctx, math.ZeroInt())
+				err = suite.poolKeeper.Distributions.Set(suite.ctx, suite.ctx.HeaderInfo().Time, math.ZeroInt())
 				suite.Require().NoError(err)
+				suite.mockStreamFunds(math.NewInt(0))
 			},
 			recipientAddress: []sdk.AccAddress{recipient},
 			expErr:           false,
@@ -504,9 +507,8 @@ func (suite *KeeperTestSuite) TestWithdrawContinuousFund() {
 				// Set recipient fund percentage and recipient fund distribution
 				err = suite.poolKeeper.RecipientFundDistribution.Set(suite.ctx, recipient, math.ZeroInt())
 				suite.Require().NoError(err)
-				toDistribute := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(100000)))
-				suite.mockStreamFunds()
-				err = suite.poolKeeper.SetToDistribute(suite.ctx, toDistribute, suite.poolKeeper.GetAuthority())
+				suite.mockStreamFunds(math.NewInt(100000))
+				err = suite.poolKeeper.SetToDistribute(suite.ctx)
 				suite.Require().NoError(err)
 			},
 			recipientAddress: []sdk.AccAddress{recipient},
@@ -530,9 +532,8 @@ func (suite *KeeperTestSuite) TestWithdrawContinuousFund() {
 				// Set recipient fund percentage and recipient fund distribution
 				err = suite.poolKeeper.RecipientFundDistribution.Set(suite.ctx, recipient, math.ZeroInt())
 				suite.Require().NoError(err)
-				toDistribute := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(100000)))
-				suite.mockStreamFunds()
-				err = suite.poolKeeper.SetToDistribute(suite.ctx, toDistribute, suite.poolKeeper.GetAuthority())
+				suite.mockStreamFunds(math.NewInt(100000))
+				err = suite.poolKeeper.SetToDistribute(suite.ctx)
 				suite.Require().NoError(err)
 			},
 			recipientAddress: []sdk.AccAddress{recipient},
@@ -588,9 +589,8 @@ func (suite *KeeperTestSuite) TestWithdrawContinuousFund() {
 				err = suite.poolKeeper.RecipientFundDistribution.Set(suite.ctx, recipient3, math.ZeroInt())
 				suite.Require().NoError(err)
 
-				toDistribute := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(100000)))
-				suite.mockStreamFunds()
-				err = suite.poolKeeper.SetToDistribute(suite.ctx, toDistribute, suite.poolKeeper.GetAuthority())
+				suite.mockStreamFunds(math.NewInt(100000))
+				err = suite.poolKeeper.SetToDistribute(suite.ctx)
 				suite.Require().NoError(err)
 			},
 			recipientAddress: []sdk.AccAddress{recipient, recipient2, recipient3},
@@ -813,11 +813,6 @@ func (suite *KeeperTestSuite) TestCancelContinuousFund() {
 			expErr:    true,
 			expErrMsg: "empty address string is not allowed",
 		},
-		"no recipient found": {
-			recipientAddr: recipientAddr,
-			expErr:        true,
-			expErrMsg:     "no recipient found to cancel continuous fund",
-		},
 		"all good with unclaimed funds for recipient": {
 			preRun: func() {
 				// Set fund 1
@@ -853,9 +848,8 @@ func (suite *KeeperTestSuite) TestCancelContinuousFund() {
 				suite.Require().NoError(err)
 
 				// Set ToDistribute
-				toDistribute := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(100000)))
-				suite.mockStreamFunds()
-				err = suite.poolKeeper.SetToDistribute(suite.ctx, toDistribute, suite.poolKeeper.GetAuthority())
+				suite.mockStreamFunds(math.NewInt(100000))
+				err = suite.poolKeeper.SetToDistribute(suite.ctx)
 				suite.Require().NoError(err)
 
 				// withdraw funds for fund request 2
@@ -960,9 +954,8 @@ func (suite *KeeperTestSuite) TestWithdrawExpiredFunds() {
 	})
 	suite.Require().NoError(err)
 
-	toDistribute := sdk.NewCoins(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(100000)))
-	suite.mockStreamFunds()
-	err = suite.poolKeeper.SetToDistribute(suite.ctx, toDistribute, suite.poolKeeper.GetAuthority())
+	suite.mockStreamFunds(math.NewInt(100000))
+	err = suite.poolKeeper.SetToDistribute(suite.ctx)
 	suite.Require().NoError(err)
 
 	suite.mockWithdrawContinuousFund()
@@ -973,11 +966,17 @@ func (suite *KeeperTestSuite) TestWithdrawExpiredFunds() {
 	header.Time = expiration.Add(1 * time.Second)
 	suite.ctx = suite.ctx.WithHeaderInfo(header)
 
-	_, err = suite.msgServer.WithdrawContinuousFund(suite.ctx, &types.MsgWithdrawContinuousFund{RecipientAddress: recipientStrAddr})
-	suite.Require().ErrorContains(err, "continuous fund expired for recipient")
+	// If we keep calling WithdrawContinuousFund, it should not error and return always an amount of 0
+	withdrawRes, err := suite.msgServer.WithdrawContinuousFund(suite.ctx, &types.MsgWithdrawContinuousFund{RecipientAddress: recipientStrAddr})
+	suite.Require().True(withdrawRes.Amount.IsZero())
+	suite.Require().NoError(err)
 
-	suite.mockStreamFunds()
-	err = suite.poolKeeper.SetToDistribute(suite.ctx, toDistribute, suite.poolKeeper.GetAuthority())
+	withdrawRes, err = suite.msgServer.WithdrawContinuousFund(suite.ctx, &types.MsgWithdrawContinuousFund{RecipientAddress: recipientStrAddr})
+	suite.Require().True(withdrawRes.Amount.IsZero())
+	suite.Require().NoError(err)
+
+	suite.mockStreamFunds(math.NewInt(100000))
+	err = suite.poolKeeper.SetToDistribute(suite.ctx)
 	suite.Require().NoError(err)
 
 	suite.mockWithdrawContinuousFund()
@@ -985,9 +984,58 @@ func (suite *KeeperTestSuite) TestWithdrawExpiredFunds() {
 	suite.Require().NoError(err)
 
 	res, err := suite.msgServer.CancelContinuousFund(suite.ctx, &types.MsgCancelContinuousFund{
+		Authority:        suite.poolKeeper.GetAuthority(),
+		RecipientAddress: recipient2StrAddr,
+	})
+	suite.Require().NoError(err)
+	suite.Require().Equal(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(0)), res.WithdrawnAllocatedFund)
+
+	// canceling an expired continuous fund, won't error
+	res, err = suite.msgServer.CancelContinuousFund(suite.ctx, &types.MsgCancelContinuousFund{
 		Authority:        suite.poolKeeper.GetAuthority(),
 		RecipientAddress: recipientStrAddr,
 	})
 	suite.Require().NoError(err)
 	suite.Require().Equal(sdk.NewCoin(sdk.DefaultBondDenom, math.NewInt(0)), res.WithdrawnAllocatedFund)
+
+	// if we try to cancel again the same continuout fund, it won't error, it will still distribute funds if needed.
+	res, err = suite.msgServer.CancelContinuousFund(suite.ctx, &types.MsgCancelContinuousFund{
+		Authority:        suite.poolKeeper.GetAuthority(),
+		RecipientAddress: recipientStrAddr,
+	})
+	suite.Require().NoError(err)
+	suite.Require().True(res.WithdrawnAllocatedFund.IsNil())
+}
+
+func (suite *KeeperTestSuite) TestFundCommunityPool() {
+	sender := []byte("fundingAddr1____________________")
+	addrCodec := codectestutil.CodecOptions{}.GetAddressCodec()
+	senderAddr, err := addrCodec.BytesToString(sender)
+	suite.Require().NoError(err)
+
+	amount := sdk.NewCoins(sdk.NewInt64Coin(sdk.DefaultBondDenom, 1000000))
+	suite.bankKeeper.EXPECT().SendCoinsFromAccountToModule(gomock.Any(), sender, types.ModuleName, amount).Return(nil).Times(1)
+
+	_, err = suite.msgServer.FundCommunityPool(suite.ctx, &types.MsgFundCommunityPool{
+		Amount:    amount,
+		Depositor: senderAddr,
+	})
+	suite.Require().NoError(err)
+}
+
+func (suite *KeeperTestSuite) TestCommunityPoolSpend() {
+	recipient := []byte("fundingAddr1____________________")
+	addrCodec := codectestutil.CodecOptions{}.GetAddressCodec()
+	recipientAddr, err := addrCodec.BytesToString(recipient)
+	suite.Require().NoError(err)
+
+	amount := sdk.NewCoins(sdk.NewInt64Coin(sdk.DefaultBondDenom, 1000000))
+	suite.bankKeeper.EXPECT().SendCoinsFromModuleToAccount(gomock.Any(), types.ModuleName, recipient, amount).Return(nil).Times(1)
+
+	_, err = suite.msgServer.CommunityPoolSpend(suite.ctx, &types.MsgCommunityPoolSpend{
+		Authority: suite.poolKeeper.GetAuthority(),
+		Recipient: recipientAddr,
+		Amount:    amount,
+	})
+	suite.Require().NoError(err)
 }

x/protocolpool/module.go

@@ -29,6 +29,7 @@ var (
 	_ appmodule.HasServices           = AppModule{}
 	_ appmodule.HasGenesis            = AppModule{}
 	_ appmodule.HasRegisterInterfaces = AppModule{}
+	_ appmodule.HasBeginBlocker       = AppModule{}
 )
 
 // AppModule implements an application module for the pool module
@@ -112,5 +113,10 @@ func (am AppModule) ExportGenesis(ctx context.Context) (json.RawMessage, error)
 	return am.cdc.MarshalJSON(gs)
 }
 
+// BeginBlock implements appmodule.HasBeginBlocker.
+func (am AppModule) BeginBlock(ctx context.Context) error {
+	return am.keeper.BeginBlocker(ctx)
+}
+
 // ConsensusVersion implements HasConsensusVersion
 func (AppModule) ConsensusVersion() uint64 { return ConsensusVersion }

x/protocolpool/proto/cosmos/protocolpool/v1/genesis.proto

@@ -6,6 +6,7 @@ option go_package = "cosmossdk.io/x/protocolpool/types";
 import "cosmos/protocolpool/v1/types.proto";
 import "gogoproto/gogo.proto";
 import "cosmos_proto/cosmos.proto";
+import "google/protobuf/timestamp.proto";
 
 // GenesisState defines the protocolpool module's genesis state.
 message GenesisState {
@@ -14,9 +15,26 @@ message GenesisState {
   // Budget defines the budget proposals at genesis.
   repeated Budget budget = 2;
 
-  string to_distribute = 3 [
+  // last_balance contains the amount of tokens yet to be distributed, will be zero if
+  // there are no funds to distribute.
+  string last_balance = 3 [
     (cosmos_proto.scalar)  = "cosmos.Int",
     (gogoproto.customtype) = "cosmossdk.io/math.Int",
     (gogoproto.nullable)   = false
   ];
+
+  // distributions contains the list of distributions to be made to continuous
+  // funds and budgets. It contains time in order to distribute to non-expired
+  // funds only.
+  repeated Distribution distributions = 4;
 }
+
+message Distribution {
+  string amount = 3 [
+    (cosmos_proto.scalar)  = "cosmos.Int",
+    (gogoproto.customtype) = "cosmossdk.io/math.Int",
+    (gogoproto.nullable)   = false
+  ];
+
+  google.protobuf.Timestamp time = 6 [(gogoproto.stdtime) = true];
+}

x/protocolpool/types/errors.go

@@ -3,6 +3,6 @@ package types
 import "cosmossdk.io/errors"
 
 var (
-	ErrInvalidSigner   = errors.Register(ModuleName, 2, "expected authority account as only signer for community pool spend message")
-	ErrNoRecipientFund = errors.Register(ModuleName, 3, "no recipient found")
+	ErrInvalidSigner    = errors.Register(ModuleName, 2, "expected authority account as only signer for community pool spend message")
+	ErrNoRecipientFound = errors.Register(ModuleName, 3, "no recipient found")
 )

x/protocolpool/types/genesis.go

@@ -13,6 +13,8 @@ func NewGenesisState(cf []*ContinuousFund, budget []*Budget) *GenesisState {
 	return &GenesisState{
 		ContinuousFund: cf,
 		Budget:         budget,
+		LastBalance:    math.ZeroInt(),
+		Distributions:  []*Distribution{},
 	}
 }
 

x/protocolpool/types/genesis.pb.go

@@ -9,15 +9,19 @@ import (
 	_ "github.com/cosmos/cosmos-proto"
 	_ "github.com/cosmos/gogoproto/gogoproto"
 	proto "github.com/cosmos/gogoproto/proto"
+	github_com_cosmos_gogoproto_types "github.com/cosmos/gogoproto/types"
+	_ "google.golang.org/protobuf/types/known/timestamppb"
 	io "io"
 	math "math"
 	math_bits "math/bits"
+	time "time"
 )
 
 // Reference imports to suppress errors if they are not otherwise used.
 var _ = proto.Marshal
 var _ = fmt.Errorf
 var _ = math.Inf
+var _ = time.Kitchen
 
 // This is a compile-time assertion to ensure that this generated file
 // is compatible with the proto package it is being compiled against.
@@ -30,8 +34,14 @@ type GenesisState struct {
 	// ContinuousFund defines the continuous funds at genesis.
 	ContinuousFund []*ContinuousFund `protobuf:"bytes,1,rep,name=continuous_fund,json=continuousFund,proto3" json:"continuous_fund,omitempty"`
 	// Budget defines the budget proposals at genesis.
-	Budget       []*Budget             `protobuf:"bytes,2,rep,name=budget,proto3" json:"budget,omitempty"`
-	ToDistribute cosmossdk_io_math.Int `protobuf:"bytes,3,opt,name=to_distribute,json=toDistribute,proto3,customtype=cosmossdk.io/math.Int" json:"to_distribute"`
+	Budget []*Budget `protobuf:"bytes,2,rep,name=budget,proto3" json:"budget,omitempty"`
+	// last_balance contains the amount of tokens yet to be distributed, will be zero if
+	// there are no funds to distribute.
+	LastBalance cosmossdk_io_math.Int `protobuf:"bytes,3,opt,name=last_balance,json=lastBalance,proto3,customtype=cosmossdk.io/math.Int" json:"last_balance"`
+	// distributions contains the list of distributions to be made to continuous
+	// funds and budgets. It contains time in order to distribute to non-expired
+	// funds only.
+	Distributions []*Distribution `protobuf:"bytes,4,rep,name=distributions,proto3" json:"distributions,omitempty"`
 }
 
 func (m *GenesisState) Reset()         { *m = GenesisState{} }
@@ -81,8 +91,61 @@ func (m *GenesisState) GetBudget() []*Budget {
 	return nil
 }
 
+func (m *GenesisState) GetDistributions() []*Distribution {
+	if m != nil {
+		return m.Distributions
+	}
+	return nil
+}
+
+type Distribution struct {
+	Amount cosmossdk_io_math.Int `protobuf:"bytes,3,opt,name=amount,proto3,customtype=cosmossdk.io/math.Int" json:"amount"`
+	Time   *time.Time            `protobuf:"bytes,6,opt,name=time,proto3,stdtime" json:"time,omitempty"`
+}
+
+func (m *Distribution) Reset()         { *m = Distribution{} }
+func (m *Distribution) String() string { return proto.CompactTextString(m) }
+func (*Distribution) ProtoMessage()    {}
+func (*Distribution) Descriptor() ([]byte, []int) {
+	return fileDescriptor_72560a99455b4146, []int{1}
+}
+func (m *Distribution) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *Distribution) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_Distribution.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *Distribution) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Distribution.Merge(m, src)
+}
+func (m *Distribution) XXX_Size() int {
+	return m.Size()
+}
+func (m *Distribution) XXX_DiscardUnknown() {
+	xxx_messageInfo_Distribution.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Distribution proto.InternalMessageInfo
+
+func (m *Distribution) GetTime() *time.Time {
+	if m != nil {
+		return m.Time
+	}
+	return nil
+}
+
 func init() {
 	proto.RegisterType((*GenesisState)(nil), "cosmos.protocolpool.v1.GenesisState")
+	proto.RegisterType((*Distribution)(nil), "cosmos.protocolpool.v1.Distribution")
 }
 
 func init() {
@@ -90,26 +153,32 @@ func init() {
 }
 
 var fileDescriptor_72560a99455b4146 = []byte{
-	// 290 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x52, 0x49, 0xce, 0x2f, 0xce,
-	0xcd, 0x2f, 0xd6, 0x2f, 0x28, 0xca, 0x2f, 0xc9, 0x4f, 0xce, 0xcf, 0x29, 0xc8, 0xcf, 0xcf, 0xd1,
-	0x2f, 0x33, 0xd4, 0x4f, 0x4f, 0xcd, 0x4b, 0x2d, 0xce, 0x2c, 0xd6, 0x03, 0x8b, 0x0b, 0x89, 0x41,
-	0x54, 0xe9, 0x21, 0xab, 0xd2, 0x2b, 0x33, 0x94, 0x52, 0xc2, 0xa1, 0xbb, 0xa4, 0xb2, 0x20, 0x15,
-	0xaa, 0x5a, 0x4a, 0x24, 0x3d, 0x3f, 0x3d, 0x1f, 0xcc, 0xd4, 0x07, 0xb1, 0xa0, 0xa2, 0x92, 0x10,
-	0x9d, 0xf1, 0x10, 0x09, 0x64, 0xe3, 0x95, 0x5e, 0x32, 0x72, 0xf1, 0xb8, 0x43, 0xac, 0x0f, 0x2e,
-	0x49, 0x2c, 0x49, 0x15, 0xf2, 0xe7, 0xe2, 0x4f, 0xce, 0xcf, 0x2b, 0xc9, 0xcc, 0x2b, 0xcd, 0x2f,
-	0x2d, 0x8e, 0x4f, 0x2b, 0xcd, 0x4b, 0x91, 0x60, 0x54, 0x60, 0xd6, 0xe0, 0x36, 0x52, 0xd3, 0xc3,
-	0xee, 0x2e, 0x3d, 0x67, 0xb8, 0x72, 0xb7, 0xd2, 0xbc, 0x94, 0x20, 0xbe, 0x64, 0x14, 0xbe, 0x90,
-	0x19, 0x17, 0x5b, 0x52, 0x69, 0x4a, 0x7a, 0x6a, 0x89, 0x04, 0x13, 0xd8, 0x1c, 0x39, 0x5c, 0xe6,
-	0x38, 0x81, 0x55, 0x05, 0x41, 0x55, 0x0b, 0x05, 0x70, 0xf1, 0x96, 0xe4, 0xc7, 0xa7, 0x64, 0x16,
-	0x97, 0x14, 0x65, 0x26, 0x95, 0x96, 0xa4, 0x4a, 0x30, 0x2b, 0x30, 0x6a, 0x70, 0x3a, 0x69, 0x9f,
-	0xb8, 0x27, 0xcf, 0x70, 0xeb, 0x9e, 0xbc, 0x28, 0xc4, 0x94, 0xe2, 0x94, 0x6c, 0xbd, 0xcc, 0x7c,
-	0xfd, 0xdc, 0xc4, 0x92, 0x0c, 0x3d, 0xcf, 0xbc, 0x92, 0x4b, 0x5b, 0x74, 0xb9, 0xa0, 0xc6, 0x7b,
-	0xe6, 0x95, 0x04, 0xf1, 0x94, 0xe4, 0xbb, 0xc0, 0x0d, 0x70, 0xb2, 0x3e, 0xf1, 0x48, 0x8e, 0xf1,
-	0xc2, 0x23, 0x39, 0xc6, 0x07, 0x8f, 0xe4, 0x18, 0x27, 0x3c, 0x96, 0x63, 0xb8, 0xf0, 0x58, 0x8e,
-	0xe1, 0xc6, 0x63, 0x39, 0x86, 0x28, 0x45, 0x14, 0xc3, 0x2a, 0x50, 0x83, 0x18, 0x1c, 0xbe, 0x49,
-	0x6c, 0x60, 0x31, 0x63, 0x40, 0x00, 0x00, 0x00, 0xff, 0xff, 0x1e, 0x8d, 0xab, 0x75, 0xc4, 0x01,
-	0x00, 0x00,
+	// 393 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x92, 0xcf, 0xca, 0xd3, 0x40,
+	0x14, 0xc5, 0x33, 0x5f, 0x4b, 0xc0, 0x69, 0x55, 0x08, 0x2a, 0x31, 0x8b, 0xa4, 0x96, 0x22, 0x05,
+	0x71, 0x42, 0xab, 0xb8, 0x71, 0x97, 0x8a, 0x52, 0x17, 0x0a, 0xd1, 0x95, 0x9b, 0x92, 0x3f, 0xd3,
+	0x18, 0x4c, 0xe6, 0x86, 0xce, 0x4c, 0xd1, 0x47, 0x70, 0xd7, 0x77, 0xd1, 0x87, 0xe8, 0xb2, 0xb8,
+	0x12, 0x17, 0x55, 0xda, 0x17, 0x91, 0xce, 0xa4, 0x92, 0x80, 0xdd, 0x7c, 0xbb, 0xc9, 0xbd, 0xbf,
+	0x73, 0x72, 0xe0, 0x5c, 0x3c, 0x4a, 0x80, 0x97, 0xc0, 0xfd, 0x6a, 0x05, 0x02, 0x12, 0x28, 0x2a,
+	0x80, 0xc2, 0x5f, 0x4f, 0xfc, 0x8c, 0x32, 0xca, 0x73, 0x4e, 0xd4, 0xdc, 0xba, 0xa7, 0x29, 0xd2,
+	0xa4, 0xc8, 0x7a, 0xe2, 0x0c, 0x2f, 0xa8, 0xc5, 0x97, 0x8a, 0xd6, 0xb4, 0x73, 0x27, 0x83, 0x0c,
+	0xd4, 0xd3, 0x3f, 0xbd, 0xea, 0xe9, 0x7d, 0xad, 0x5c, 0xe8, 0x45, 0xd3, 0xde, 0xf1, 0x32, 0x80,
+	0xac, 0xa0, 0xda, 0x34, 0x96, 0x4b, 0x5f, 0xe4, 0x25, 0xe5, 0x22, 0x2a, 0x2b, 0x0d, 0x0c, 0xbf,
+	0x5d, 0xe1, 0xfe, 0x2b, 0x9d, 0xef, 0x9d, 0x88, 0x04, 0xb5, 0xde, 0xe2, 0xdb, 0x09, 0x30, 0x91,
+	0x33, 0x09, 0x92, 0x2f, 0x96, 0x92, 0xa5, 0x36, 0x1a, 0x74, 0xc6, 0xbd, 0xe9, 0x43, 0xf2, 0xff,
+	0xe0, 0x64, 0xf6, 0x0f, 0x7f, 0x29, 0x59, 0x1a, 0xde, 0x4a, 0x5a, 0xdf, 0xd6, 0x33, 0x6c, 0xc6,
+	0x32, 0xcd, 0xa8, 0xb0, 0xaf, 0x94, 0x8f, 0x7b, 0xc9, 0x27, 0x50, 0x54, 0x58, 0xd3, 0xd6, 0x1b,
+	0xdc, 0x2f, 0x22, 0x2e, 0x16, 0x71, 0x54, 0x44, 0x2c, 0xa1, 0x76, 0x67, 0x80, 0xc6, 0x37, 0x82,
+	0x47, 0xdb, 0xbd, 0x67, 0xfc, 0xda, 0x7b, 0x77, 0xb5, 0x09, 0x4f, 0x3f, 0x91, 0x1c, 0xfc, 0x32,
+	0x12, 0x1f, 0xc9, 0x9c, 0x89, 0x1f, 0xdf, 0x1f, 0xe3, 0xda, 0x7d, 0xce, 0x44, 0xd8, 0x3b, 0x19,
+	0x04, 0x5a, 0x6f, 0xbd, 0xc6, 0x37, 0xd3, 0x9c, 0x8b, 0x55, 0x1e, 0x4b, 0x91, 0x03, 0xe3, 0x76,
+	0x57, 0xc5, 0x19, 0x5d, 0x8a, 0xf3, 0xa2, 0x01, 0x87, 0x6d, 0xe9, 0xf0, 0x2b, 0xc2, 0xfd, 0xe6,
+	0xde, 0x9a, 0x61, 0x33, 0x2a, 0x41, 0x32, 0x71, 0x9d, 0x98, 0xb5, 0xd4, 0x7a, 0x8a, 0xbb, 0xa7,
+	0x7a, 0x6c, 0x73, 0x80, 0xc6, 0xbd, 0xa9, 0x43, 0x74, 0x77, 0xe4, 0xdc, 0x1d, 0x79, 0x7f, 0xee,
+	0x2e, 0xe8, 0x6e, 0x7e, 0x7b, 0x28, 0x54, 0x74, 0xf0, 0x7c, 0x7b, 0x70, 0xd1, 0xee, 0xe0, 0xa2,
+	0x3f, 0x07, 0x17, 0x6d, 0x8e, 0xae, 0xb1, 0x3b, 0xba, 0xc6, 0xcf, 0xa3, 0x6b, 0x7c, 0x78, 0xd0,
+	0xfa, 0xf9, 0xe7, 0xf6, 0x65, 0xa9, 0xb3, 0x8a, 0x4d, 0x35, 0x7b, 0xf2, 0x37, 0x00, 0x00, 0xff,
+	0xff, 0x0c, 0x77, 0xdd, 0xf7, 0xbb, 0x02, 0x00, 0x00,
 }
 
 func (m *GenesisState) Marshal() (dAtA []byte, err error) {
@@ -132,10 +201,24 @@ func (m *GenesisState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if len(m.Distributions) > 0 {
+		for iNdEx := len(m.Distributions) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.Distributions[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintGenesis(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x22
+		}
+	}
 	{
-		size := m.ToDistribute.Size()
+		size := m.LastBalance.Size()
 		i -= size
-		if _, err := m.ToDistribute.MarshalTo(dAtA[i:]); err != nil {
+		if _, err := m.LastBalance.MarshalTo(dAtA[i:]); err != nil {
 			return 0, err
 		}
 		i = encodeVarintGenesis(dAtA, i, uint64(size))
@@ -173,6 +256,49 @@ func (m *GenesisState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *Distribution) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *Distribution) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *Distribution) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.Time != nil {
+		n1, err1 := github_com_cosmos_gogoproto_types.StdTimeMarshalTo(*m.Time, dAtA[i-github_com_cosmos_gogoproto_types.SizeOfStdTime(*m.Time):])
+		if err1 != nil {
+			return 0, err1
+		}
+		i -= n1
+		i = encodeVarintGenesis(dAtA, i, uint64(n1))
+		i--
+		dAtA[i] = 0x32
+	}
+	{
+		size := m.Amount.Size()
+		i -= size
+		if _, err := m.Amount.MarshalTo(dAtA[i:]); err != nil {
+			return 0, err
+		}
+		i = encodeVarintGenesis(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0x1a
+	return len(dAtA) - i, nil
+}
+
 func encodeVarintGenesis(dAtA []byte, offset int, v uint64) int {
 	offset -= sovGenesis(v)
 	base := offset
@@ -202,8 +328,29 @@ func (m *GenesisState) Size() (n int) {
 			n += 1 + l + sovGenesis(uint64(l))
 		}
 	}
-	l = m.ToDistribute.Size()
+	l = m.LastBalance.Size()
 	n += 1 + l + sovGenesis(uint64(l))
+	if len(m.Distributions) > 0 {
+		for _, e := range m.Distributions {
+			l = e.Size()
+			n += 1 + l + sovGenesis(uint64(l))
+		}
+	}
+	return n
+}
+
+func (m *Distribution) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = m.Amount.Size()
+	n += 1 + l + sovGenesis(uint64(l))
+	if m.Time != nil {
+		l = github_com_cosmos_gogoproto_types.SizeOfStdTime(*m.Time)
+		n += 1 + l + sovGenesis(uint64(l))
+	}
 	return n
 }
 
@@ -312,7 +459,7 @@ func (m *GenesisState) Unmarshal(dAtA []byte) error {
 			iNdEx = postIndex
 		case 3:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ToDistribute", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field LastBalance", wireType)
 			}
 			var stringLen uint64
 			for shift := uint(0); ; shift += 7 {
@@ -340,7 +487,161 @@ func (m *GenesisState) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			if err := m.ToDistribute.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			if err := m.LastBalance.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Distributions", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenesis
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthGenesis
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenesis
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Distributions = append(m.Distributions, &Distribution{})
+			if err := m.Distributions[len(m.Distributions)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipGenesis(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthGenesis
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *Distribution) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowGenesis
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: Distribution: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: Distribution: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Amount", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenesis
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthGenesis
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenesis
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.Amount.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 6:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Time", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenesis
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthGenesis
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenesis
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Time == nil {
+				m.Time = new(time.Time)
+			}
+			if err := github_com_cosmos_gogoproto_types.StdTimeUnmarshal(m.Time, dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex

x/protocolpool/types/keys.go

@@ -9,6 +9,9 @@ const (
 	// StreamAccount is the name constant used for stream account
 	StreamAccount = "stream_acc"
 
+	// ProtocolPoolDistrAccount is an intermediary account that holds the funds to be distributed to the protocolpool accounts.
+	ProtocolPoolDistrAccount = "protocolpool_distr"
+
 	// StoreKey is the store key string for protocolpool
 	StoreKey = ModuleName
 
@@ -26,5 +29,6 @@ var (
 	ContinuousFundKey            = collections.NewPrefix(3)
 	RecipientFundPercentageKey   = collections.NewPrefix(4)
 	RecipientFundDistributionKey = collections.NewPrefix(5)
-	ToDistributeKey              = collections.NewPrefix(6)
+	DistributionsKey             = collections.NewPrefix(6)
+	LastBalanceKey               = collections.NewPrefix(7)
 )