Genesis export OOM

x/wasm/alias.go

@@ -52,6 +52,7 @@ var (
 	DefaultParams             = types.DefaultParams
 	InitGenesis               = keeper.InitGenesis
 	ExportGenesis             = keeper.ExportGenesis
+	ExportGenesisStream       = keeper.ExportGenesisStream
 	NewMessageHandler         = keeper.NewDefaultMessageHandler
 	DefaultEncoders           = keeper.DefaultEncoders
 	EncodeBankMsg             = keeper.EncodeBankMsg

x/wasm/keeper/genesis.go

@@ -1,6 +1,8 @@
 package keeper
 
 import (
+	"fmt"
+
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 	abci "github.com/tendermint/tendermint/abci/types"
@@ -34,6 +36,7 @@
 			}
 		}
 	}
+	fmt.Println("Got maxCodeID = ", maxCodeID)
 
 	var maxContractID int
 	for i, contract := range data.Contracts {
@@ -47,16 +50,21 @@
 		}
 		maxContractID = i + 1 // not ideal but max(contractID) is not persisted otherwise
 	}
+	fmt.Println("maxContractID = ", maxContractID)
 
+	fmt.Println("Iterating over data.Sequences = ", data.Sequences)
 	for i, seq := range data.Sequences {
+		fmt.Println("i = ", i, ", seq = ", seq)
 		err := keeper.importAutoIncrementID(ctx, seq.IDKey, seq.Value)
 		if err != nil {
 			return nil, sdkerrors.Wrapf(err, "sequence number %d", i)
 		}
 	}
 
 	// sanity check seq values
+	fmt.Println("types.KeyLastCodeID = ", string(types.KeyLastCodeID))
 	seqVal := keeper.PeekAutoIncrementID(ctx, types.KeyLastCodeID)
+	fmt.Println("seqVal = ", seqVal)
 	if seqVal <= maxCodeID {
 		return nil, sdkerrors.Wrapf(types.ErrInvalid, "seq %s with value: %d must be greater than: %d ", string(types.KeyLastCodeID), seqVal, maxCodeID)
 	}
@@ -126,3 +134,78 @@
 
 	return &genState
 }
+
+const GENSIS_STATE_STREAM_BUF_THRESHOLD = 50000
+
+func ExportGenesisStream(ctx sdk.Context, keeper *Keeper) <-chan *types.GenesisState {
+	ch := make(chan *types.GenesisState)
+	go func() {
+		var genState types.GenesisState
+		genState.Params = keeper.GetParams(ctx)
+		ch <- &genState
+
+		// Needs to be first because there are invariant checks when importing that need sequences info
+		for _, k := range [][]byte{types.KeyLastCodeID, types.KeyLastInstanceID} {
+			var genState types.GenesisState
+			genState.Params = keeper.GetParams(ctx)
+			genState.Sequences = append(genState.Sequences, types.Sequence{
+				IDKey: k,
+				Value: keeper.PeekAutoIncrementID(ctx, k),
+			})
+			ch <- &genState
+		}
+
+		keeper.IterateCodeInfos(ctx, func(codeID uint64, info types.CodeInfo) bool {
+			var genState types.GenesisState
+			genState.Params = keeper.GetParams(ctx)
+			bytecode, err := keeper.GetByteCode(ctx, codeID)
+			if err != nil {
+				panic(err)
+			}
+			genState.Codes = append(genState.Codes, types.Code{
+				CodeID:    codeID,
+				CodeInfo:  info,
+				CodeBytes: bytecode,
+				Pinned:    keeper.IsPinnedCode(ctx, codeID),
+			})
+			ch <- &genState
+			return false
+		})
+
+		fmt.Println("About to IterateContractInfo")
+		keeper.IterateContractInfo(ctx, func(addr sdk.AccAddress, contract types.ContractInfo) bool {
+			// redact contract info
+			contract.Created = nil
+			var state []types.Model
+			keeper.IterateContractState(ctx, addr, func(key, value []byte) bool {
+				state = append(state, types.Model{Key: key, Value: value})
+				if len(state) > GENSIS_STATE_STREAM_BUF_THRESHOLD {
+					var genState types.GenesisState
+					genState.Params = keeper.GetParams(ctx)
+					genState.Contracts = append(genState.Contracts, types.Contract{
+						ContractAddress: addr.String(),
+						ContractInfo:    contract,
+						ContractState:   state,
+					})
+					ch <- &genState
+					state = nil
+				}
+				return false
+			})
+			// flush any remaining state
+			var genState types.GenesisState
+			genState.Params = keeper.GetParams(ctx)
+			genState.Contracts = append(genState.Contracts, types.Contract{
+				ContractAddress: addr.String(),
+				ContractInfo:    contract,
+				ContractState:   state,
+			})
+			ch <- &genState
+			return false
+		})
+		fmt.Println("Done with IterateContractInfo")
+
+		close(ch)
+	}()
+	return ch
+}

x/wasm/keeper/keeper.go

@@ -164,6 +164,7 @@ func (k Keeper) GetParams(ctx sdk.Context) types.Params {
 }
 
 func (k Keeper) SetParams(ctx sdk.Context, ps types.Params) {
+	fmt.Println("In x/wasm, keeper: trying to set params = ", ps)
 	k.paramSpace.SetParamSet(ctx, &ps)
 }
 
@@ -1000,6 +1001,7 @@ func (k Keeper) PeekAutoIncrementID(ctx sdk.Context, lastIDKey []byte) uint64 {
 
 func (k Keeper) importAutoIncrementID(ctx sdk.Context, lastIDKey []byte, val uint64) error {
 	store := ctx.KVStore(k.storeKey)
+	fmt.Println("lastIDKey = ", string(lastIDKey))
 	if store.Has(lastIDKey) {
 		return sdkerrors.Wrapf(types.ErrDuplicate, "autoincrement id: %s", string(lastIDKey))
 	}

x/wasm/module.go

@@ -3,6 +3,7 @@
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"math/rand"
 
 	"github.com/cosmos/cosmos-sdk/client"
@@ -76,6 +77,33 @@
 	return ValidateGenesis(data)
 }
 
+func (am AppModuleBasic) ValidateGenesisStream(cdc codec.JSONCodec, config client.TxEncodingConfig, genesisCh <-chan json.RawMessage) error {
+	fmt.Println("In wasm ValidateGenesisStream")
+	var err error
+	doneCh := make(chan struct{})
+	genesisStateCh := make(chan GenesisState)
+	go func() {
+		err = types.ValidateGenesisStream(genesisStateCh)
+		doneCh <- struct{}{}
+	}()
+	go func() {
+		defer close(genesisStateCh)
+		for genesis := range genesisCh {
+			var data GenesisState
+			err_ := cdc.UnmarshalJSON(genesis, &data)
+			if err_ != nil {
+				err = err_
+				doneCh <- struct{}{}
+				return
+			}
+			genesisStateCh <- data
+		}
+	}()
+	<-doneCh
+	fmt.Println("done with wasm ValidateGenesisStream err = ", err)
+	return err
+}
+
 // RegisterRESTRoutes registers the REST routes for the wasm module.
 func (AppModuleBasic) RegisterRESTRoutes(cliCtx client.Context, rtr *mux.Router) {
 	rest.RegisterRoutes(cliCtx, rtr)
@@ -170,7 +198,20 @@
 // module.
 func (am AppModule) ExportGenesis(ctx sdk.Context, cdc codec.JSONCodec) json.RawMessage {
 	gs := ExportGenesis(ctx, am.keeper)
-	return cdc.MustMarshalJSON(gs)
+	marshalled := cdc.MustMarshalJSON(gs)
+	return marshalled
+}
+
+func (am AppModule) StreamGenesis(ctx sdk.Context, cdc codec.JSONCodec) <-chan json.RawMessage {
+	ch := ExportGenesisStream(ctx, am.keeper)
+	chRaw := make(chan json.RawMessage)
+	go func() {
+		for genState := range ch {
+			chRaw <- cdc.MustMarshalJSON(genState)
+		}
+		close(chRaw)
+	}()
+	return chRaw
 }
 
 // BeginBlock returns the begin blocker for the wasm module.

x/wasm/types/genesis.go

@@ -40,6 +40,34 @@ func (s GenesisState) ValidateBasic() error {
 	return nil
 }
 
+func (s GenesisState) ValidateBasicStream(dataCh chan GenesisState) error {
+	// just need one of them to return true for this one
+	if err := s.Params.ValidateBasic(); err != nil {
+		return sdkerrors.Wrap(err, "params")
+	}
+	for i := range s.Codes {
+		if err := s.Codes[i].ValidateBasic(); err != nil {
+			return sdkerrors.Wrapf(err, "code: %d", i)
+		}
+	}
+	for i := range s.Contracts {
+		if err := s.Contracts[i].ValidateBasic(); err != nil {
+			return sdkerrors.Wrapf(err, "contract: %d", i)
+		}
+	}
+	for i := range s.Sequences {
+		if err := s.Sequences[i].ValidateBasic(); err != nil {
+			return sdkerrors.Wrapf(err, "sequence: %d", i)
+		}
+	}
+	for i := range s.GenMsgs {
+		if err := s.GenMsgs[i].ValidateBasic(); err != nil {
+			return sdkerrors.Wrapf(err, "gen message: %d", i)
+		}
+	}
+	return nil
+}
+
 func (c Code) ValidateBasic() error {
 	if c.CodeID == 0 {
 		return sdkerrors.Wrap(ErrEmpty, "code id")
@@ -100,6 +128,49 @@ func ValidateGenesis(data GenesisState) error {
 	return data.ValidateBasic()
 }
 
+// ValidateGenesisStream performs basic validation of wasm genesis data over a stream
+// of wasm genesis states. It needs to pass the params validation at least one chunk and
+// other checks on every chunk of the stream.
+func ValidateGenesisStream(genesisStateCh <-chan GenesisState) error {
+	passedParamsCheck := false
+	var paramCheckErr error
+	var otherErr error
+	for s := range genesisStateCh {
+		if otherErr != nil {
+			continue
+		}
+		if err := s.Params.ValidateBasic(); err != nil {
+			paramCheckErr = sdkerrors.Wrap(err, "params")
+		} else {
+			passedParamsCheck = true
+		}
+		for i := range s.Codes {
+			if err := s.Codes[i].ValidateBasic(); err != nil {
+				otherErr = sdkerrors.Wrapf(err, "code: %d", i)
+			}
+		}
+		for i := range s.Contracts {
+			if err := s.Contracts[i].ValidateBasic(); err != nil {
+				otherErr = sdkerrors.Wrapf(err, "contract: %d", i)
+			}
+		}
+		for i := range s.Sequences {
+			if err := s.Sequences[i].ValidateBasic(); err != nil {
+				otherErr = sdkerrors.Wrapf(err, "sequence: %d", i)
+			}
+		}
+		for i := range s.GenMsgs {
+			if err := s.GenMsgs[i].ValidateBasic(); err != nil {
+				otherErr = sdkerrors.Wrapf(err, "gen message: %d", i)
+			}
+		}
+	}
+	if !passedParamsCheck {
+		return paramCheckErr
+	}
+	return otherErr
+}
+
 var _ codectypes.UnpackInterfacesMessage = GenesisState{}
 
 // UnpackInterfaces implements codectypes.UnpackInterfaces