diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0821a48a1bd65..0d43ba15394c2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -55,6 +55,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 * [\#9576](https://github.com/cosmos/cosmos-sdk/pull/9576) Add debug error message to query result when enabled
 * (types) [\#11200](https://github.com/cosmos/cosmos-sdk/pull/11200) Added `Min()` and `Max()` operations on sdk.Coins.
+* [#11267](https://github.com/cosmos/cosmos-sdk/pull/11267) Add hooks to allow app modules to add things to state-sync (backport #10961).
 
 ## [v0.45.1](https://github.com/cosmos/cosmos-sdk/releases/tag/v0.45.1) - 2022-02-03
 
diff --git a/baseapp/baseapp.go b/baseapp/baseapp.go
index 3a73f6ef20d9b..023610bc55af5 100644
--- a/baseapp/baseapp.go
+++ b/baseapp/baseapp.go
@@ -273,6 +273,22 @@ func DefaultStoreLoader(ms sdk.CommitMultiStore) error {
 	return ms.LoadLatestVersion()
 }
 
+// CommitMultiStore returns the root multi-store.
+// App constructor can use this to access the `cms`.
+// UNSAFE: only safe to use during app initialization.
+func (app *BaseApp) CommitMultiStore() sdk.CommitMultiStore {
+	if app.sealed {
+		panic("cannot call CommitMultiStore() after baseapp is sealed")
+	}
+	return app.cms
+}
+
+// SnapshotManager returns the snapshot manager.
+// application use this to register extra extension snapshotters.
+func (app *BaseApp) SnapshotManager() *snapshots.Manager {
+	return app.snapshotManager
+}
+
 // LoadVersion loads the BaseApp application version. It will panic if called
 // more than once on a running baseapp.
 func (app *BaseApp) LoadVersion(version int64) error {
diff --git a/proto/cosmos/base/snapshots/v1beta1/snapshot.proto b/proto/cosmos/base/snapshots/v1beta1/snapshot.proto
index 5dba369fd4484..1d2365bef6520 100644
--- a/proto/cosmos/base/snapshots/v1beta1/snapshot.proto
+++ b/proto/cosmos/base/snapshots/v1beta1/snapshot.proto
@@ -20,8 +20,6 @@ message Metadata {
 }
 
 // SnapshotItem is an item contained in a rootmulti.Store snapshot.
-//
-// Since: cosmos-sdk 0.46
 message SnapshotItem {
   // item is the specific type of snapshot item.
   oneof item {
@@ -29,56 +27,31 @@ message SnapshotItem {
     SnapshotIAVLItem         iavl              = 2 [(gogoproto.customname) = "IAVL"];
     SnapshotExtensionMeta    extension         = 3;
     SnapshotExtensionPayload extension_payload = 4;
-    SnapshotKVItem           kv                = 5 [(gogoproto.customname) = "KV"];
-    SnapshotSchema           schema            = 6;
   }
 }
 
 // SnapshotStoreItem contains metadata about a snapshotted store.
-//
-// Since: cosmos-sdk 0.46
 message SnapshotStoreItem {
   string name = 1;
 }
 
 // SnapshotIAVLItem is an exported IAVL node.
-//
-// Since: cosmos-sdk 0.46
 message SnapshotIAVLItem {
-  bytes key   = 1;
-  bytes value = 2;
+  bytes key     = 1;
+  bytes value   = 2;
   // version is block height
   int64 version = 3;
   // height is depth of the tree.
-  int32 height = 4;
+  int32 height  = 4;
 }
 
 // SnapshotExtensionMeta contains metadata about an external snapshotter.
-//
-// Since: cosmos-sdk 0.46
 message SnapshotExtensionMeta {
   string name   = 1;
   uint32 format = 2;
 }
 
 // SnapshotExtensionPayload contains payloads of an external snapshotter.
-//
-// Since: cosmos-sdk 0.46
 message SnapshotExtensionPayload {
   bytes payload = 1;
 }
-
-// SnapshotKVItem is an exported Key/Value Pair
-//
-// Since: cosmos-sdk 0.46
-message SnapshotKVItem {
-  bytes key   = 1;
-  bytes value = 2;
-}
-
-// SnapshotSchema is an exported schema of smt store
-//
-// Since: cosmos-sdk 0.46
-message SnapshotSchema {
-  repeated bytes keys = 1;
-}
diff --git a/server/mock/store.go b/server/mock/store.go
index 8094980932f26..360bb8371877f 100644
--- a/server/mock/store.go
+++ b/server/mock/store.go
@@ -6,16 +6,15 @@ import (
 	protoio "github.com/gogo/protobuf/io"
 	dbm "github.com/tendermint/tm-db"
 
-	pruningtypes "github.com/cosmos/cosmos-sdk/pruning/types"
 	snapshottypes "github.com/cosmos/cosmos-sdk/snapshots/types"
-	storetypes "github.com/cosmos/cosmos-sdk/store/types"
+	store "github.com/cosmos/cosmos-sdk/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
 var _ sdk.MultiStore = multiStore{}
 
 type multiStore struct {
-	kv map[storetypes.StoreKey]kvStore
+	kv map[sdk.StoreKey]kvStore
 }
 
 func (ms multiStore) CacheMultiStore() sdk.CacheMultiStore {
@@ -26,15 +25,15 @@ func (ms multiStore) CacheMultiStoreWithVersion(_ int64) (sdk.CacheMultiStore, e
 	panic("not implemented")
 }
 
-func (ms multiStore) CacheWrap() storetypes.CacheWrap {
+func (ms multiStore) CacheWrap() sdk.CacheWrap {
 	panic("not implemented")
 }
 
-func (ms multiStore) CacheWrapWithTrace(_ io.Writer, _ sdk.TraceContext) storetypes.CacheWrap {
+func (ms multiStore) CacheWrapWithTrace(_ io.Writer, _ sdk.TraceContext) sdk.CacheWrap {
 	panic("not implemented")
 }
 
-func (ms multiStore) CacheWrapWithListeners(_ storetypes.StoreKey, _ []storetypes.WriteListener) storetypes.CacheWrap {
+func (ms multiStore) CacheWrapWithListeners(_ store.StoreKey, _ []store.WriteListener) store.CacheWrap {
 	panic("not implemented")
 }
 
@@ -50,39 +49,39 @@ func (ms multiStore) SetTracer(w io.Writer) sdk.MultiStore {
 	panic("not implemented")
 }
 
-func (ms multiStore) AddListeners(key storetypes.StoreKey, listeners []storetypes.WriteListener) {
+func (ms multiStore) AddListeners(key store.StoreKey, listeners []store.WriteListener) {
 	panic("not implemented")
 }
 
-func (ms multiStore) ListeningEnabled(key storetypes.StoreKey) bool {
+func (ms multiStore) ListeningEnabled(key store.StoreKey) bool {
 	panic("not implemented")
 }
 
-func (ms multiStore) Commit() storetypes.CommitID {
+func (ms multiStore) Commit() sdk.CommitID {
 	panic("not implemented")
 }
 
-func (ms multiStore) LastCommitID() storetypes.CommitID {
+func (ms multiStore) LastCommitID() sdk.CommitID {
 	panic("not implemented")
 }
 
-func (ms multiStore) SetPruning(opts pruningtypes.PruningOptions) {
+func (ms multiStore) SetPruning(opts sdk.PruningOptions) {
 	panic("not implemented")
 }
 
-func (ms multiStore) GetPruning() pruningtypes.PruningOptions {
+func (ms multiStore) GetPruning() sdk.PruningOptions {
 	panic("not implemented")
 }
 
-func (ms multiStore) GetCommitKVStore(key storetypes.StoreKey) storetypes.CommitKVStore {
+func (ms multiStore) GetCommitKVStore(key sdk.StoreKey) sdk.CommitKVStore {
 	panic("not implemented")
 }
 
-func (ms multiStore) GetCommitStore(key storetypes.StoreKey) storetypes.CommitStore {
+func (ms multiStore) GetCommitStore(key sdk.StoreKey) sdk.CommitStore {
 	panic("not implemented")
 }
 
-func (ms multiStore) MountStoreWithDB(key storetypes.StoreKey, typ storetypes.StoreType, db dbm.DB) {
+func (ms multiStore) MountStoreWithDB(key sdk.StoreKey, typ sdk.StoreType, db dbm.DB) {
 	ms.kv[key] = kvStore{store: make(map[string][]byte)}
 }
 
@@ -90,11 +89,11 @@ func (ms multiStore) LoadLatestVersion() error {
 	return nil
 }
 
-func (ms multiStore) LoadLatestVersionAndUpgrade(upgrades *storetypes.StoreUpgrades) error {
+func (ms multiStore) LoadLatestVersionAndUpgrade(upgrades *store.StoreUpgrades) error {
 	return nil
 }
 
-func (ms multiStore) LoadVersionAndUpgrade(ver int64, upgrades *storetypes.StoreUpgrades) error {
+func (ms multiStore) LoadVersionAndUpgrade(ver int64, upgrades *store.StoreUpgrades) error {
 	panic("not implemented")
 }
 
@@ -102,23 +101,15 @@ func (ms multiStore) LoadVersion(ver int64) error {
 	panic("not implemented")
 }
 
-func (ms multiStore) GetKVStore(key storetypes.StoreKey) sdk.KVStore {
+func (ms multiStore) GetKVStore(key sdk.StoreKey) sdk.KVStore {
 	return ms.kv[key]
 }
 
-func (ms multiStore) GetStore(key storetypes.StoreKey) sdk.Store {
+func (ms multiStore) GetStore(key sdk.StoreKey) sdk.Store {
 	panic("not implemented")
 }
 
-func (ms multiStore) GetStoreType() storetypes.StoreType {
-	panic("not implemented")
-}
-
-func (ms multiStore) PruneSnapshotHeight(height int64) {
-	panic("not implemented")
-}
-
-func (ms multiStore) SetSnapshotInterval(snapshotInterval uint64) {
+func (ms multiStore) GetStoreType() sdk.StoreType {
 	panic("not implemented")
 }
 
@@ -129,10 +120,6 @@ func (ms multiStore) SetIAVLCacheSize(size int) {
 	panic("not implemented")
 }
 
-func (ms multiStore) SetIAVLCacheSize(size int) {
-	panic("not implemented")
-}
-
 func (ms multiStore) SetInitialVersion(version int64) error {
 	panic("not implemented")
 }
@@ -153,19 +140,19 @@ type kvStore struct {
 	store map[string][]byte
 }
 
-func (kv kvStore) CacheWrap() storetypes.CacheWrap {
+func (kv kvStore) CacheWrap() sdk.CacheWrap {
 	panic("not implemented")
 }
 
-func (kv kvStore) CacheWrapWithTrace(w io.Writer, tc sdk.TraceContext) storetypes.CacheWrap {
+func (kv kvStore) CacheWrapWithTrace(w io.Writer, tc sdk.TraceContext) sdk.CacheWrap {
 	panic("not implemented")
 }
 
-func (kv kvStore) CacheWrapWithListeners(_ storetypes.StoreKey, _ []storetypes.WriteListener) storetypes.CacheWrap {
+func (kv kvStore) CacheWrapWithListeners(_ store.StoreKey, _ []store.WriteListener) store.CacheWrap {
 	panic("not implemented")
 }
 
-func (kv kvStore) GetStoreType() storetypes.StoreType {
+func (kv kvStore) GetStoreType() sdk.StoreType {
 	panic("not implemented")
 }
 
@@ -183,7 +170,7 @@ func (kv kvStore) Has(key []byte) bool {
 }
 
 func (kv kvStore) Set(key, value []byte) {
-	storetypes.AssertValidKey(key)
+	store.AssertValidKey(key)
 	kv.store[string(key)] = value
 }
 
@@ -216,5 +203,5 @@ func (kv kvStore) ReverseSubspaceIterator(prefix []byte) sdk.Iterator {
 }
 
 func NewCommitMultiStore() sdk.CommitMultiStore {
-	return multiStore{kv: make(map[storetypes.StoreKey]kvStore)}
+	return multiStore{kv: make(map[sdk.StoreKey]kvStore)}
 }
diff --git a/snapshots/helpers_test.go b/snapshots/helpers_test.go
index 2dc18102c3cfa..8c9a67b1b0e53 100644
--- a/snapshots/helpers_test.go
+++ b/snapshots/helpers_test.go
@@ -7,18 +7,18 @@ import (
 	"crypto/sha256"
 	"errors"
 	"io"
+	"io/ioutil"
+	"os"
 	"testing"
 	"time"
 
 	protoio "github.com/gogo/protobuf/io"
 	"github.com/stretchr/testify/require"
-	"github.com/tendermint/tendermint/libs/log"
 	db "github.com/tendermint/tm-db"
 
 	"github.com/cosmos/cosmos-sdk/snapshots"
+	"github.com/cosmos/cosmos-sdk/snapshots/types"
 	snapshottypes "github.com/cosmos/cosmos-sdk/snapshots/types"
-	"github.com/cosmos/cosmos-sdk/testutil"
-	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 )
 
@@ -44,7 +44,7 @@ func hash(chunks [][]byte) []byte {
 func makeChunks(chunks [][]byte) <-chan io.ReadCloser {
 	ch := make(chan io.ReadCloser, len(chunks))
 	for _, chunk := range chunks {
-		ch <- io.NopCloser(bytes.NewReader(chunk))
+		ch <- ioutil.NopCloser(bytes.NewReader(chunk))
 	}
 	close(ch)
 	return ch
@@ -53,7 +53,7 @@ func makeChunks(chunks [][]byte) <-chan io.ReadCloser {
 func readChunks(chunks <-chan io.ReadCloser) [][]byte {
 	bodies := [][]byte{}
 	for chunk := range chunks {
-		body, err := io.ReadAll(chunk)
+		body, err := ioutil.ReadAll(chunk)
 		if err != nil {
 			panic(err)
 		}
@@ -63,7 +63,7 @@ func readChunks(chunks <-chan io.ReadCloser) [][]byte {
 }
 
 // snapshotItems serialize a array of bytes as SnapshotItem_ExtensionPayload, and return the chunks.
-func snapshotItems(items [][]byte, ext snapshottypes.ExtensionSnapshotter) [][]byte {
+func snapshotItems(items [][]byte) [][]byte {
 	// copy the same parameters from the code
 	snapshotChunkSize := uint64(10e6)
 	snapshotBufferSize := int(snapshotChunkSize)
@@ -75,24 +75,12 @@ func snapshotItems(items [][]byte, ext snapshottypes.ExtensionSnapshotter) [][]b
 		zWriter, _ := zlib.NewWriterLevel(bufWriter, 7)
 		protoWriter := protoio.NewDelimitedWriter(zWriter)
 		for _, item := range items {
-			_ = snapshottypes.WriteExtensionPayload(protoWriter, item)
+			types.WriteExtensionItem(protoWriter, item)
 		}
-		// write extension metadata
-		_ = protoWriter.WriteMsg(&snapshottypes.SnapshotItem{
-			Item: &snapshottypes.SnapshotItem_Extension{
-				Extension: &snapshottypes.SnapshotExtensionMeta{
-					Name:   ext.SnapshotName(),
-					Format: ext.SnapshotFormat(),
-				},
-			},
-		})
-		_ = ext.SnapshotExtension(0, func(payload []byte) error {
-			return snapshottypes.WriteExtensionPayload(protoWriter, payload)
-		})
-		_ = protoWriter.Close()
-		_ = zWriter.Close()
-		_ = bufWriter.Flush()
-		_ = chunkWriter.Close()
+		protoWriter.Close()
+		zWriter.Close()
+		bufWriter.Flush()
+		chunkWriter.Close()
 	}()
 
 	var chunks [][]byte
@@ -103,31 +91,27 @@ func snapshotItems(items [][]byte, ext snapshottypes.ExtensionSnapshotter) [][]b
 		}
 		chunks = append(chunks, chunk)
 	}
-
 	return chunks
 }
 
 type mockSnapshotter struct {
-	items            [][]byte
-	prunedHeights    map[int64]struct{}
-	snapshotInterval uint64
+	items [][]byte
 }
 
 func (m *mockSnapshotter) Restore(
 	height uint64, format uint32, protoReader protoio.Reader,
 ) (snapshottypes.SnapshotItem, error) {
 	if format == 0 {
-		return snapshottypes.SnapshotItem{}, snapshottypes.ErrUnknownFormat
+		return snapshottypes.SnapshotItem{}, types.ErrUnknownFormat
 	}
 	if m.items != nil {
 		return snapshottypes.SnapshotItem{}, errors.New("already has contents")
 	}
 
-	var item snapshottypes.SnapshotItem
 	m.items = [][]byte{}
 	for {
-		item.Reset()
-		err := protoReader.ReadMsg(&item)
+		item := &snapshottypes.SnapshotItem{}
+		err := protoReader.ReadMsg(item)
 		if err == io.EOF {
 			break
 		} else if err != nil {
@@ -135,17 +119,17 @@ func (m *mockSnapshotter) Restore(
 		}
 		payload := item.GetExtensionPayload()
 		if payload == nil {
-			break
+			return snapshottypes.SnapshotItem{}, sdkerrors.Wrap(err, "invalid protobuf message")
 		}
 		m.items = append(m.items, payload.Payload)
 	}
 
-	return item, nil
+	return snapshottypes.SnapshotItem{}, nil
 }
 
 func (m *mockSnapshotter) Snapshot(height uint64, protoWriter protoio.Writer) error {
 	for _, item := range m.items {
-		if err := snapshottypes.WriteExtensionPayload(protoWriter, item); err != nil {
+		if err := types.WriteExtensionItem(protoWriter, item); err != nil {
 			return err
 		}
 	}
@@ -153,40 +137,30 @@ func (m *mockSnapshotter) Snapshot(height uint64, protoWriter protoio.Writer) er
 }
 
 func (m *mockSnapshotter) SnapshotFormat() uint32 {
-	return snapshottypes.CurrentFormat
+	return 1
 }
-
 func (m *mockSnapshotter) SupportedFormats() []uint32 {
-	return []uint32{snapshottypes.CurrentFormat}
-}
-
-func (m *mockSnapshotter) PruneSnapshotHeight(height int64) {
-	m.prunedHeights[height] = struct{}{}
-}
-
-func (m *mockSnapshotter) GetSnapshotInterval() uint64 {
-	return m.snapshotInterval
-}
-
-func (m *mockSnapshotter) SetSnapshotInterval(snapshotInterval uint64) {
-	m.snapshotInterval = snapshotInterval
+	return []uint32{1}
 }
 
 // setupBusyManager creates a manager with an empty store that is busy creating a snapshot at height 1.
 // The snapshot will complete when the returned closer is called.
 func setupBusyManager(t *testing.T) *snapshots.Manager {
-	store, err := snapshots.NewStore(db.NewMemDB(), testutil.GetTempDir(t))
+	// ioutil.TempDir() is used instead of testing.T.TempDir()
+	// see https://github.com/cosmos/cosmos-sdk/pull/8475 for
+	// this change's rationale.
+	tempdir, err := ioutil.TempDir("", "")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = os.RemoveAll(tempdir) })
+
+	store, err := snapshots.NewStore(db.NewMemDB(), tempdir)
 	require.NoError(t, err)
 	hung := newHungSnapshotter()
-	hung.SetSnapshotInterval(opts.Interval)
-	mgr := snapshots.NewManager(store, opts, hung, nil, log.NewNopLogger())
-	require.Equal(t, opts.Interval, hung.snapshotInterval)
+	mgr := snapshots.NewManager(store, hung)
 
 	go func() {
 		_, err := mgr.Create(1)
 		require.NoError(t, err)
-		_, didPruneHeight := hung.prunedHeights[1]
-		require.True(t, didPruneHeight)
 	}()
 	time.Sleep(10 * time.Millisecond)
 	t.Cleanup(hung.Close)
@@ -196,15 +170,12 @@ func setupBusyManager(t *testing.T) *snapshots.Manager {
 
 // hungSnapshotter can be used to test operations in progress. Call close to end the snapshot.
 type hungSnapshotter struct {
-	ch               chan struct{}
-	prunedHeights    map[int64]struct{}
-	snapshotInterval uint64
+	ch chan struct{}
 }
 
 func newHungSnapshotter() *hungSnapshotter {
 	return &hungSnapshotter{
-		ch:            make(chan struct{}),
-		prunedHeights: make(map[int64]struct{}),
+		ch: make(chan struct{}),
 	}
 }
 
@@ -217,65 +188,8 @@ func (m *hungSnapshotter) Snapshot(height uint64, protoWriter protoio.Writer) er
 	return nil
 }
 
-func (m *hungSnapshotter) PruneSnapshotHeight(height int64) {
-	m.prunedHeights[height] = struct{}{}
-}
-
-func (m *hungSnapshotter) SetSnapshotInterval(snapshotInterval uint64) {
-	m.snapshotInterval = snapshotInterval
-}
-
 func (m *hungSnapshotter) Restore(
 	height uint64, format uint32, protoReader protoio.Reader,
 ) (snapshottypes.SnapshotItem, error) {
 	panic("not implemented")
 }
-
-type extSnapshotter struct {
-	state []uint64
-}
-
-func newExtSnapshotter(count int) *extSnapshotter {
-	state := make([]uint64, 0, count)
-	for i := 0; i < count; i++ {
-		state = append(state, uint64(i))
-	}
-	return &extSnapshotter{
-		state,
-	}
-}
-
-func (s *extSnapshotter) SnapshotName() string {
-	return "mock"
-}
-
-func (s *extSnapshotter) SnapshotFormat() uint32 {
-	return 1
-}
-
-func (s *extSnapshotter) SupportedFormats() []uint32 {
-	return []uint32{1}
-}
-
-func (s *extSnapshotter) SnapshotExtension(height uint64, payloadWriter snapshottypes.ExtensionPayloadWriter) error {
-	for _, i := range s.state {
-		if err := payloadWriter(sdk.Uint64ToBigEndian(uint64(i))); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (s *extSnapshotter) RestoreExtension(height uint64, format uint32, payloadReader snapshottypes.ExtensionPayloadReader) error {
-	for {
-		payload, err := payloadReader()
-		if err == io.EOF {
-			break
-		} else if err != nil {
-			return err
-		}
-		s.state = append(s.state, sdk.BigEndianToUint64(payload))
-	}
-	// finalize restoration
-	return nil
-}
diff --git a/snapshots/manager.go b/snapshots/manager.go
index efc123e9e4981..02878963d4915 100644
--- a/snapshots/manager.go
+++ b/snapshots/manager.go
@@ -3,19 +3,37 @@ package snapshots
 import (
 	"bytes"
 	"crypto/sha256"
-	"errors"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"sort"
 	"sync"
 
-	"github.com/tendermint/tendermint/libs/log"
-
 	"github.com/cosmos/cosmos-sdk/snapshots/types"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 )
 
+const (
+	opNone     operation = ""
+	opSnapshot operation = "snapshot"
+	opPrune    operation = "prune"
+	opRestore  operation = "restore"
+
+	chunkBufferSize = 4
+
+	snapshotMaxItemSize = int(64e6) // SDK has no key/value size limit, so we set an arbitrary limit
+)
+
+// operation represents a Manager operation. Only one operation can be in progress at a time.
+type operation string
+
+// restoreDone represents the result of a restore operation.
+type restoreDone struct {
+	complete bool  // if true, restore completed successfully (not prematurely)
+	err      error // if non-nil, restore errored
+}
+
 // Manager manages snapshot and restore operations for an app, making sure only a single
 // long-running operation is in progress at any given time, and provides convenience methods
 // mirroring the ABCI interface.
@@ -23,20 +41,16 @@ import (
 // Although the ABCI interface (and this manager) passes chunks as byte slices, the internal
 // snapshot/restore APIs use IO streams (i.e. chan io.ReadCloser), for two reasons:
 //
-//  1. In the future, ABCI should support streaming. Consider e.g. InitChain during chain
-//     upgrades, which currently passes the entire chain state as an in-memory byte slice.
-//     https://github.com/tendermint/tendermint/issues/5184
+// 1) In the future, ABCI should support streaming. Consider e.g. InitChain during chain
+//    upgrades, which currently passes the entire chain state as an in-memory byte slice.
+//    https://github.com/tendermint/tendermint/issues/5184
 //
-//  2. io.ReadCloser streams automatically propagate IO errors, and can pass arbitrary
-//     errors via io.Pipe.CloseWithError().
+// 2) io.ReadCloser streams automatically propagate IO errors, and can pass arbitrary
+//    errors via io.Pipe.CloseWithError().
 type Manager struct {
-	extensions map[string]types.ExtensionSnapshotter
-	// store is the snapshot store where all completed snapshots are persisted.
-	store *Store
-	opts  types.SnapshotOptions
-	// multistore is the store from which snapshots are taken.
+	store      *Store
 	multistore types.Snapshotter
-	logger     log.Logger
+	extensions map[string]types.ExtensionSnapshotter
 
 	mtx                sync.Mutex
 	operation          operation
@@ -46,47 +60,26 @@ type Manager struct {
 	restoreChunkIndex  uint32
 }
 
-// operation represents a Manager operation. Only one operation can be in progress at a time.
-type operation string
-
-// restoreDone represents the result of a restore operation.
-type restoreDone struct {
-	complete bool  // if true, restore completed successfully (not prematurely)
-	err      error // if non-nil, restore errored
-}
-
-const (
-	opNone     operation = ""
-	opSnapshot operation = "snapshot"
-	opPrune    operation = "prune"
-	opRestore  operation = "restore"
-
-	chunkBufferSize = 4
-
-	snapshotMaxItemSize = int(64e6) // SDK has no key/value size limit, so we set an arbitrary limit
-)
-
-var ErrOptsZeroSnapshotInterval = errors.New("snaphot-interval must not be 0")
-
 // NewManager creates a new manager.
-func NewManager(store *Store, opts types.SnapshotOptions, multistore types.Snapshotter, extensions map[string]types.ExtensionSnapshotter, logger log.Logger) *Manager {
-	if extensions == nil {
-		extensions = map[string]types.ExtensionSnapshotter{}
+func NewManager(store *Store, multistore types.Snapshotter) *Manager {
+	return &Manager{
+		store:      store,
+		multistore: multistore,
+		extensions: make(map[string]types.ExtensionSnapshotter),
 	}
+}
+
+// NewManagerWithExtensions creates a new manager.
+func NewManagerWithExtensions(store *Store, multistore types.Snapshotter, extensions map[string]types.ExtensionSnapshotter) *Manager {
 	return &Manager{
 		store:      store,
-		opts:       opts,
 		multistore: multistore,
 		extensions: extensions,
-		logger:     logger,
 	}
 }
 
 // RegisterExtensions register extension snapshotters to manager
 func (m *Manager) RegisterExtensions(extensions ...types.ExtensionSnapshotter) error {
-	if m.extensions == nil {
-		m.extensions = make(map[string]types.ExtensionSnapshotter, len(extensions))
-	}
 	for _, extension := range extensions {
 		name := extension.SnapshotName()
 		if _, ok := m.extensions[name]; ok {
@@ -138,22 +131,15 @@ func (m *Manager) endLocked() {
 	m.restoreChunkIndex = 0
 }
 
-// GetInterval returns snapshot interval represented in heights.
-func (m *Manager) GetInterval() uint64 {
-	return m.opts.Interval
-}
-
-// GetKeepRecent returns snapshot keep-recent represented in heights.
-func (m *Manager) GetKeepRecent() uint32 {
-	return m.opts.KeepRecent
-}
+// sortedExtensionNames sort extension names for deterministic iteration.
+func (m *Manager) sortedExtensionNames() []string {
+	names := make([]string, 0, len(m.extensions))
+	for name := range m.extensions {
+		names = append(names, name)
+	}
 
-// GetSnapshotBlockRetentionHeights returns the number of heights needed
-// for block retention. Blocks since the oldest available snapshot must be
-// available for state sync nodes to catch up (oldest because a node may be
-// restoring an old snapshot while a new snapshot was taken).
-func (m *Manager) GetSnapshotBlockRetentionHeights() int64 {
-	return int64(m.opts.Interval * uint64(m.opts.KeepRecent))
+	sort.Strings(names)
+	return names
 }
 
 // Create creates a snapshot and returns its metadata.
@@ -161,9 +147,6 @@ func (m *Manager) Create(height uint64) (*types.Snapshot, error) {
 	if m == nil {
 		return nil, sdkerrors.Wrap(sdkerrors.ErrLogic, "no snapshot store configured")
 	}
-
-	defer m.multistore.PruneSnapshotHeight(int64(height))
-
 	err := m.begin(opSnapshot)
 	if err != nil {
 		return nil, err
@@ -193,12 +176,7 @@ func (m *Manager) createSnapshot(height uint64, ch chan<- io.ReadCloser) {
 	if streamWriter == nil {
 		return
 	}
-	defer func() {
-		if err := streamWriter.Close(); err != nil {
-			streamWriter.CloseWithError(err)
-		}
-	}()
-
+	defer streamWriter.Close()
 	if err := m.multistore.Snapshot(height, streamWriter); err != nil {
 		streamWriter.CloseWithError(err)
 		return
@@ -218,10 +196,7 @@ func (m *Manager) createSnapshot(height uint64, ch chan<- io.ReadCloser) {
 			streamWriter.CloseWithError(err)
 			return
 		}
-		payloadWriter := func(payload []byte) error {
-			return types.WriteExtensionPayload(streamWriter, payload)
-		}
-		if err := extension.SnapshotExtension(height, payloadWriter); err != nil {
+		if err := extension.Snapshot(height, streamWriter); err != nil {
 			streamWriter.CloseWithError(err)
 			return
 		}
@@ -245,7 +220,7 @@ func (m *Manager) LoadChunk(height uint64, format uint32, chunk uint32) ([]byte,
 	}
 	defer reader.Close()
 
-	return io.ReadAll(reader)
+	return ioutil.ReadAll(reader)
 }
 
 // Prune prunes snapshots, if no other operations are in progress.
@@ -311,40 +286,24 @@ func (m *Manager) Restore(snapshot types.Snapshot) error {
 
 // restoreSnapshot do the heavy work of snapshot restoration after preliminary checks on request have passed.
 func (m *Manager) restoreSnapshot(snapshot types.Snapshot, chChunks <-chan io.ReadCloser) error {
-	var nextItem types.SnapshotItem
-
 	streamReader, err := NewStreamReader(chChunks)
 	if err != nil {
 		return err
 	}
 	defer streamReader.Close()
 
-	// payloadReader reads an extension payload for extension snapshotter, it returns `io.EOF` at extension boundaries.
-	payloadReader := func() ([]byte, error) {
-		nextItem.Reset()
-		if err := streamReader.ReadMsg(&nextItem); err != nil {
-			return nil, err
-		}
-		payload := nextItem.GetExtensionPayload()
-		if payload == nil {
-			return nil, io.EOF
-		}
-		return payload.Payload, nil
-	}
-
-	nextItem, err = m.multistore.Restore(snapshot.Height, snapshot.Format, streamReader)
+	next, err := m.multistore.Restore(snapshot.Height, snapshot.Format, streamReader)
 	if err != nil {
 		return sdkerrors.Wrap(err, "multistore restore")
 	}
-
 	for {
-		if nextItem.Item == nil {
+		if next.Item == nil {
 			// end of stream
 			break
 		}
-		metadata := nextItem.GetExtension()
+		metadata := next.GetExtension()
 		if metadata == nil {
-			return sdkerrors.Wrapf(sdkerrors.ErrLogic, "unknown snapshot item %T", nextItem.Item)
+			return sdkerrors.Wrapf(sdkerrors.ErrLogic, "unknown snapshot item %T", next.Item)
 		}
 		extension, ok := m.extensions[metadata.Name]
 		if !ok {
@@ -353,14 +312,10 @@ func (m *Manager) restoreSnapshot(snapshot types.Snapshot, chChunks <-chan io.Re
 		if !IsFormatSupported(extension, metadata.Format) {
 			return sdkerrors.Wrapf(types.ErrUnknownFormat, "format %v for extension %s", metadata.Format, metadata.Name)
 		}
-
-		if err := extension.RestoreExtension(snapshot.Height, metadata.Format, payloadReader); err != nil {
+		next, err = extension.Restore(snapshot.Height, metadata.Format, streamReader)
+		if err != nil {
 			return sdkerrors.Wrapf(err, "extension %s restore", metadata.Name)
 		}
-
-		if nextItem.GetExtensionPayload() != nil {
-			return sdkerrors.Wrapf(err, "extension %s don't exhausted payload stream", metadata.Name)
-		}
 	}
 	return nil
 }
@@ -398,7 +353,7 @@ func (m *Manager) RestoreChunk(chunk []byte) (bool, error) {
 	}
 
 	// Pass the chunk to the restore, and wait for completion if it was the final one.
-	m.chRestore <- io.NopCloser(bytes.NewReader(chunk))
+	m.chRestore <- ioutil.NopCloser(bytes.NewReader(chunk))
 	m.restoreChunkIndex++
 
 	if int(m.restoreChunkIndex) >= len(m.restoreChunkHashes) {
@@ -417,17 +372,6 @@ func (m *Manager) RestoreChunk(chunk []byte) (bool, error) {
 	return false, nil
 }
 
-// sortedExtensionNames sort extension names for deterministic iteration.
-func (m *Manager) sortedExtensionNames() []string {
-	names := make([]string, 0, len(m.extensions))
-	for name := range m.extensions {
-		names = append(names, name)
-	}
-
-	sort.Strings(names)
-	return names
-}
-
 // IsFormatSupported returns if the snapshotter supports restoration from given format.
 func IsFormatSupported(snapshotter types.ExtensionSnapshotter, format uint32) bool {
 	for _, i := range snapshotter.SupportedFormats() {
@@ -437,50 +381,3 @@ func IsFormatSupported(snapshotter types.ExtensionSnapshotter, format uint32) bo
 	}
 	return false
 }
-
-// SnapshotIfApplicable takes a snapshot of the current state if we are on a snapshot height.
-// It also prunes any old snapshots.
-func (m *Manager) SnapshotIfApplicable(height int64) {
-	if m == nil {
-		return
-	}
-	if !m.shouldTakeSnapshot(height) {
-		m.logger.Debug("snapshot is skipped", "height", height)
-		return
-	}
-	m.snapshot(height)
-}
-
-// shouldTakeSnapshot returns true is snapshot should be taken at height.
-func (m *Manager) shouldTakeSnapshot(height int64) bool {
-	return m.opts.Interval > 0 && uint64(height)%m.opts.Interval == 0
-}
-
-func (m *Manager) snapshot(height int64) {
-	m.logger.Info("creating state snapshot", "height", height)
-
-	if height <= 0 {
-		m.logger.Error("snapshot height must be positive", "height", height)
-		return
-	}
-
-	snapshot, err := m.Create(uint64(height))
-	if err != nil {
-		m.logger.Error("failed to create state snapshot", "height", height, "err", err)
-		return
-	}
-
-	m.logger.Info("completed state snapshot", "height", height, "format", snapshot.Format)
-
-	if m.opts.KeepRecent > 0 {
-		m.logger.Debug("pruning state snapshots")
-
-		pruned, err := m.Prune(m.opts.KeepRecent)
-		if err != nil {
-			m.logger.Error("Failed to prune state snapshots", "err", err)
-			return
-		}
-
-		m.logger.Debug("pruned state snapshots", "pruned", pruned)
-	}
-}
diff --git a/snapshots/manager_test.go b/snapshots/manager_test.go
index 58a302e87ca83..256b5b2cc8feb 100644
--- a/snapshots/manager_test.go
+++ b/snapshots/manager_test.go
@@ -6,20 +6,14 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/tendermint/tendermint/libs/log"
 
 	"github.com/cosmos/cosmos-sdk/snapshots"
 	"github.com/cosmos/cosmos-sdk/snapshots/types"
 )
 
-var opts = types.NewSnapshotOptions(1500, 2)
-
 func TestManager_List(t *testing.T) {
 	store := setupStore(t)
-	snapshotter := &mockSnapshotter{}
-	snapshotter.SetSnapshotInterval(opts.Interval)
-	manager := snapshots.NewManager(store, opts, snapshotter, nil, log.NewNopLogger())
-	require.Equal(t, opts.Interval, snapshotter.GetSnapshotInterval())
+	manager := snapshots.NewManager(store, nil)
 
 	mgrList, err := manager.List()
 	require.NoError(t, err)
@@ -38,7 +32,7 @@ func TestManager_List(t *testing.T) {
 
 func TestManager_LoadChunk(t *testing.T) {
 	store := setupStore(t)
-	manager := snapshots.NewManager(store, opts, &mockSnapshotter{}, nil, log.NewNopLogger())
+	manager := snapshots.NewManager(store, nil)
 
 	// Existing chunk should return body
 	chunk, err := manager.LoadChunk(2, 1, 1)
@@ -65,37 +59,27 @@ func TestManager_Take(t *testing.T) {
 		{7, 8, 9},
 	}
 	snapshotter := &mockSnapshotter{
-		items:         items,
-		prunedHeights: make(map[int64]struct{}),
+		items: items,
 	}
-	extSnapshotter := newExtSnapshotter(10)
-
-	expectChunks := snapshotItems(items, extSnapshotter)
-	manager := snapshots.NewManager(store, opts, snapshotter, nil, log.NewNopLogger())
-	err := manager.RegisterExtensions(extSnapshotter)
-	require.NoError(t, err)
+	expectChunks := snapshotItems(items)
+	manager := snapshots.NewManager(store, snapshotter)
 
 	// nil manager should return error
-	_, err = (*snapshots.Manager)(nil).Create(1)
+	_, err := (*snapshots.Manager)(nil).Create(1)
 	require.Error(t, err)
 
 	// creating a snapshot at a lower height than the latest should error
 	_, err = manager.Create(3)
 	require.Error(t, err)
-	_, didPruneHeight := snapshotter.prunedHeights[3]
-	require.True(t, didPruneHeight)
 
 	// creating a snapshot at a higher height should be fine, and should return it
 	snapshot, err := manager.Create(5)
 	require.NoError(t, err)
-	_, didPruneHeight = snapshotter.prunedHeights[5]
-	require.True(t, didPruneHeight)
-
 	assert.Equal(t, &types.Snapshot{
 		Height: 5,
 		Format: snapshotter.SnapshotFormat(),
 		Chunks: 1,
-		Hash:   []uint8{0x89, 0xfa, 0x18, 0xbc, 0x5a, 0xe3, 0xdc, 0x36, 0xa6, 0x95, 0x5, 0x17, 0xf9, 0x2, 0x1a, 0x55, 0x36, 0x16, 0x5d, 0x4b, 0x8b, 0x2b, 0x3d, 0xfd, 0xe, 0x2f, 0xb6, 0x40, 0x6b, 0xc3, 0xbc, 0x23},
+		Hash:   []uint8{0xcd, 0x17, 0x9e, 0x7f, 0x28, 0xb6, 0x82, 0x90, 0xc7, 0x25, 0xf3, 0x42, 0xac, 0x65, 0x73, 0x50, 0xaa, 0xa0, 0x10, 0x5c, 0x40, 0x8c, 0xd5, 0x1, 0xed, 0x82, 0xb5, 0xca, 0x8b, 0xe0, 0x83, 0xa2},
 		Metadata: types.Metadata{
 			ChunkHashes: checksums(expectChunks),
 		},
@@ -114,9 +98,7 @@ func TestManager_Take(t *testing.T) {
 
 func TestManager_Prune(t *testing.T) {
 	store := setupStore(t)
-	snapshotter := &mockSnapshotter{}
-	snapshotter.SetSnapshotInterval(opts.Interval)
-	manager := snapshots.NewManager(store, opts, snapshotter, nil, log.NewNopLogger())
+	manager := snapshots.NewManager(store, nil)
 
 	pruned, err := manager.Prune(2)
 	require.NoError(t, err)
@@ -134,13 +116,8 @@ func TestManager_Prune(t *testing.T) {
 
 func TestManager_Restore(t *testing.T) {
 	store := setupStore(t)
-	target := &mockSnapshotter{
-		prunedHeights: make(map[int64]struct{}),
-	}
-	extSnapshotter := newExtSnapshotter(0)
-	manager := snapshots.NewManager(store, opts, target, nil, log.NewNopLogger())
-	err := manager.RegisterExtensions(extSnapshotter)
-	require.NoError(t, err)
+	target := &mockSnapshotter{}
+	manager := snapshots.NewManager(store, target)
 
 	expectItems := [][]byte{
 		{1, 2, 3},
@@ -148,10 +125,10 @@ func TestManager_Restore(t *testing.T) {
 		{7, 8, 9},
 	}
 
-	chunks := snapshotItems(expectItems, newExtSnapshotter(10))
+	chunks := snapshotItems(expectItems)
 
 	// Restore errors on invalid format
-	err = manager.Restore(types.Snapshot{
+	err := manager.Restore(types.Snapshot{
 		Height:   3,
 		Format:   0,
 		Hash:     []byte{1, 2, 3},
@@ -162,13 +139,13 @@ func TestManager_Restore(t *testing.T) {
 	require.ErrorIs(t, err, types.ErrUnknownFormat)
 
 	// Restore errors on no chunks
-	err = manager.Restore(types.Snapshot{Height: 3, Format: types.CurrentFormat, Hash: []byte{1, 2, 3}})
+	err = manager.Restore(types.Snapshot{Height: 3, Format: 1, Hash: []byte{1, 2, 3}})
 	require.Error(t, err)
 
 	// Restore errors on chunk and chunkhashes mismatch
 	err = manager.Restore(types.Snapshot{
 		Height:   3,
-		Format:   types.CurrentFormat,
+		Format:   1,
 		Hash:     []byte{1, 2, 3},
 		Chunks:   4,
 		Metadata: types.Metadata{ChunkHashes: checksums(chunks)},
@@ -178,7 +155,7 @@ func TestManager_Restore(t *testing.T) {
 	// Starting a restore works
 	err = manager.Restore(types.Snapshot{
 		Height:   3,
-		Format:   types.CurrentFormat,
+		Format:   1,
 		Hash:     []byte{1, 2, 3},
 		Chunks:   1,
 		Metadata: types.Metadata{ChunkHashes: checksums(chunks)},
@@ -188,8 +165,6 @@ func TestManager_Restore(t *testing.T) {
 	// While the restore is in progress, any other operations fail
 	_, err = manager.Create(4)
 	require.Error(t, err)
-	_, didPruneHeight := target.prunedHeights[4]
-	require.True(t, didPruneHeight)
 
 	_, err = manager.Prune(1)
 	require.Error(t, err)
@@ -211,12 +186,11 @@ func TestManager_Restore(t *testing.T) {
 	}
 
 	assert.Equal(t, expectItems, target.items)
-	assert.Equal(t, 10, len(extSnapshotter.state))
 
 	// Starting a new restore should fail now, because the target already has contents.
 	err = manager.Restore(types.Snapshot{
 		Height:   3,
-		Format:   types.CurrentFormat,
+		Format:   1,
 		Hash:     []byte{1, 2, 3},
 		Chunks:   3,
 		Metadata: types.Metadata{ChunkHashes: checksums(chunks)},
@@ -229,7 +203,7 @@ func TestManager_Restore(t *testing.T) {
 	target.items = nil
 	err = manager.Restore(types.Snapshot{
 		Height:   3,
-		Format:   types.CurrentFormat,
+		Format:   1,
 		Hash:     []byte{1, 2, 3},
 		Chunks:   1,
 		Metadata: types.Metadata{ChunkHashes: checksums(chunks)},
diff --git a/snapshots/stream.go b/snapshots/stream.go
index 935028313bb55..80cd5c3dfdcb7 100644
--- a/snapshots/stream.go
+++ b/snapshots/stream.go
@@ -103,15 +103,7 @@ func (sr *StreamReader) ReadMsg(msg proto.Message) error {
 
 // Close implements io.Closer interface
 func (sr *StreamReader) Close() error {
-	var err error
-	if err1 := sr.protoReader.Close(); err1 != nil {
-		err = err1
-	}
-	if err2 := sr.zReader.Close(); err2 != nil {
-		err = err2
-	}
-	if err3 := sr.chunkReader.Close(); err3 != nil {
-		err = err3
-	}
-	return err
+	sr.protoReader.Close()
+	sr.zReader.Close()
+	return sr.chunkReader.Close()
 }
diff --git a/snapshots/types/snapshot.pb.go b/snapshots/types/snapshot.pb.go
index 703e9769e9cb0..5a70e5eb8d4da 100644
--- a/snapshots/types/snapshot.pb.go
+++ b/snapshots/types/snapshot.pb.go
@@ -146,8 +146,6 @@ func (m *Metadata) GetChunkHashes() [][]byte {
 }
 
 // SnapshotItem is an item contained in a rootmulti.Store snapshot.
-//
-// Since: cosmos-sdk 0.46
 type SnapshotItem struct {
 	// item is the specific type of snapshot item.
 	//
@@ -156,8 +154,6 @@ type SnapshotItem struct {
 	//	*SnapshotItem_IAVL
 	//	*SnapshotItem_Extension
 	//	*SnapshotItem_ExtensionPayload
-	//	*SnapshotItem_KV
-	//	*SnapshotItem_Schema
 	Item isSnapshotItem_Item `protobuf_oneof:"item"`
 }
 
@@ -212,19 +208,11 @@ type SnapshotItem_Extension struct {
 type SnapshotItem_ExtensionPayload struct {
 	ExtensionPayload *SnapshotExtensionPayload `protobuf:"bytes,4,opt,name=extension_payload,json=extensionPayload,proto3,oneof" json:"extension_payload,omitempty"`
 }
-type SnapshotItem_KV struct {
-	KV *SnapshotKVItem `protobuf:"bytes,5,opt,name=kv,proto3,oneof" json:"kv,omitempty"`
-}
-type SnapshotItem_Schema struct {
-	Schema *SnapshotSchema `protobuf:"bytes,6,opt,name=schema,proto3,oneof" json:"schema,omitempty"`
-}
 
 func (*SnapshotItem_Store) isSnapshotItem_Item()            {}
 func (*SnapshotItem_IAVL) isSnapshotItem_Item()             {}
 func (*SnapshotItem_Extension) isSnapshotItem_Item()        {}
 func (*SnapshotItem_ExtensionPayload) isSnapshotItem_Item() {}
-func (*SnapshotItem_KV) isSnapshotItem_Item()               {}
-func (*SnapshotItem_Schema) isSnapshotItem_Item()           {}
 
 func (m *SnapshotItem) GetItem() isSnapshotItem_Item {
 	if m != nil {
@@ -261,20 +249,6 @@ func (m *SnapshotItem) GetExtensionPayload() *SnapshotExtensionPayload {
 	return nil
 }
 
-func (m *SnapshotItem) GetKV() *SnapshotKVItem {
-	if x, ok := m.GetItem().(*SnapshotItem_KV); ok {
-		return x.KV
-	}
-	return nil
-}
-
-func (m *SnapshotItem) GetSchema() *SnapshotSchema {
-	if x, ok := m.GetItem().(*SnapshotItem_Schema); ok {
-		return x.Schema
-	}
-	return nil
-}
-
 // XXX_OneofWrappers is for the internal use of the proto package.
 func (*SnapshotItem) XXX_OneofWrappers() []interface{} {
 	return []interface{}{
@@ -282,14 +256,10 @@ func (*SnapshotItem) XXX_OneofWrappers() []interface{} {
 		(*SnapshotItem_IAVL)(nil),
 		(*SnapshotItem_Extension)(nil),
 		(*SnapshotItem_ExtensionPayload)(nil),
-		(*SnapshotItem_KV)(nil),
-		(*SnapshotItem_Schema)(nil),
 	}
 }
 
 // SnapshotStoreItem contains metadata about a snapshotted store.
-//
-// Since: cosmos-sdk 0.46
 type SnapshotStoreItem struct {
 	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
 }
@@ -335,15 +305,11 @@ func (m *SnapshotStoreItem) GetName() string {
 }
 
 // SnapshotIAVLItem is an exported IAVL node.
-//
-// Since: cosmos-sdk 0.46
 type SnapshotIAVLItem struct {
-	Key   []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
-	Value []byte `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
-	// version is block height
-	Version int64 `protobuf:"varint,3,opt,name=version,proto3" json:"version,omitempty"`
-	// height is depth of the tree.
-	Height int32 `protobuf:"varint,4,opt,name=height,proto3" json:"height,omitempty"`
+	Key     []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
+	Value   []byte `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
+	Version int64  `protobuf:"varint,3,opt,name=version,proto3" json:"version,omitempty"`
+	Height  int32  `protobuf:"varint,4,opt,name=height,proto3" json:"height,omitempty"`
 }
 
 func (m *SnapshotIAVLItem) Reset()         { *m = SnapshotIAVLItem{} }
@@ -408,8 +374,6 @@ func (m *SnapshotIAVLItem) GetHeight() int32 {
 }
 
 // SnapshotExtensionMeta contains metadata about an external snapshotter.
-//
-// Since: cosmos-sdk 0.46
 type SnapshotExtensionMeta struct {
 	Name   string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
 	Format uint32 `protobuf:"varint,2,opt,name=format,proto3" json:"format,omitempty"`
@@ -463,8 +427,6 @@ func (m *SnapshotExtensionMeta) GetFormat() uint32 {
 }
 
 // SnapshotExtensionPayload contains payloads of an external snapshotter.
-//
-// Since: cosmos-sdk 0.46
 type SnapshotExtensionPayload struct {
 	Payload []byte `protobuf:"bytes,1,opt,name=payload,proto3" json:"payload,omitempty"`
 }
@@ -509,108 +471,6 @@ func (m *SnapshotExtensionPayload) GetPayload() []byte {
 	return nil
 }
 
-// SnapshotKVItem is an exported Key/Value Pair
-//
-// Since: cosmos-sdk 0.46
-type SnapshotKVItem struct {
-	Key   []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
-	Value []byte `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
-}
-
-func (m *SnapshotKVItem) Reset()         { *m = SnapshotKVItem{} }
-func (m *SnapshotKVItem) String() string { return proto.CompactTextString(m) }
-func (*SnapshotKVItem) ProtoMessage()    {}
-func (*SnapshotKVItem) Descriptor() ([]byte, []int) {
-	return fileDescriptor_dd7a3c9b0a19e1ee, []int{7}
-}
-func (m *SnapshotKVItem) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *SnapshotKVItem) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	if deterministic {
-		return xxx_messageInfo_SnapshotKVItem.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 *SnapshotKVItem) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_SnapshotKVItem.Merge(m, src)
-}
-func (m *SnapshotKVItem) XXX_Size() int {
-	return m.Size()
-}
-func (m *SnapshotKVItem) XXX_DiscardUnknown() {
-	xxx_messageInfo_SnapshotKVItem.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_SnapshotKVItem proto.InternalMessageInfo
-
-func (m *SnapshotKVItem) GetKey() []byte {
-	if m != nil {
-		return m.Key
-	}
-	return nil
-}
-
-func (m *SnapshotKVItem) GetValue() []byte {
-	if m != nil {
-		return m.Value
-	}
-	return nil
-}
-
-// SnapshotSchema is an exported schema of smt store
-//
-// Since: cosmos-sdk 0.46
-type SnapshotSchema struct {
-	Keys [][]byte `protobuf:"bytes,1,rep,name=keys,proto3" json:"keys,omitempty"`
-}
-
-func (m *SnapshotSchema) Reset()         { *m = SnapshotSchema{} }
-func (m *SnapshotSchema) String() string { return proto.CompactTextString(m) }
-func (*SnapshotSchema) ProtoMessage()    {}
-func (*SnapshotSchema) Descriptor() ([]byte, []int) {
-	return fileDescriptor_dd7a3c9b0a19e1ee, []int{8}
-}
-func (m *SnapshotSchema) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *SnapshotSchema) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	if deterministic {
-		return xxx_messageInfo_SnapshotSchema.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 *SnapshotSchema) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_SnapshotSchema.Merge(m, src)
-}
-func (m *SnapshotSchema) XXX_Size() int {
-	return m.Size()
-}
-func (m *SnapshotSchema) XXX_DiscardUnknown() {
-	xxx_messageInfo_SnapshotSchema.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_SnapshotSchema proto.InternalMessageInfo
-
-func (m *SnapshotSchema) GetKeys() [][]byte {
-	if m != nil {
-		return m.Keys
-	}
-	return nil
-}
-
 func init() {
 	proto.RegisterType((*Snapshot)(nil), "cosmos.base.snapshots.v1beta1.Snapshot")
 	proto.RegisterType((*Metadata)(nil), "cosmos.base.snapshots.v1beta1.Metadata")
@@ -619,8 +479,6 @@ func init() {
 	proto.RegisterType((*SnapshotIAVLItem)(nil), "cosmos.base.snapshots.v1beta1.SnapshotIAVLItem")
 	proto.RegisterType((*SnapshotExtensionMeta)(nil), "cosmos.base.snapshots.v1beta1.SnapshotExtensionMeta")
 	proto.RegisterType((*SnapshotExtensionPayload)(nil), "cosmos.base.snapshots.v1beta1.SnapshotExtensionPayload")
-	proto.RegisterType((*SnapshotKVItem)(nil), "cosmos.base.snapshots.v1beta1.SnapshotKVItem")
-	proto.RegisterType((*SnapshotSchema)(nil), "cosmos.base.snapshots.v1beta1.SnapshotSchema")
 }
 
 func init() {
@@ -628,44 +486,40 @@ func init() {
 }
 
 var fileDescriptor_dd7a3c9b0a19e1ee = []byte{
-	// 581 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x54, 0xc1, 0x6e, 0xd3, 0x40,
-	0x10, 0xb5, 0x13, 0xc7, 0xa4, 0x63, 0x83, 0xda, 0x55, 0x41, 0x16, 0x12, 0x6e, 0xb0, 0x90, 0x9a,
-	0x43, 0x6b, 0xd3, 0x50, 0x09, 0xae, 0x04, 0x41, 0x5d, 0x15, 0x04, 0xda, 0xa0, 0x1c, 0xb8, 0x54,
-	0x9b, 0x64, 0x1b, 0x47, 0x8e, 0xb3, 0x51, 0x76, 0x63, 0x91, 0x23, 0x7f, 0xc0, 0xaf, 0xf0, 0x17,
-	0x3d, 0xf6, 0xc8, 0x29, 0x42, 0xc9, 0x8f, 0xa0, 0x5d, 0xdb, 0x69, 0x29, 0x2d, 0xa4, 0xa7, 0xcc,
-	0x4c, 0xde, 0x7b, 0x9e, 0x9d, 0xb7, 0xb3, 0xb0, 0xd7, 0x65, 0x3c, 0x61, 0x3c, 0xe8, 0x10, 0x4e,
-	0x03, 0x3e, 0x22, 0x63, 0x1e, 0x31, 0xc1, 0x83, 0xf4, 0xa0, 0x43, 0x05, 0x39, 0x58, 0x55, 0xfc,
-	0xf1, 0x84, 0x09, 0x86, 0x9e, 0x64, 0x68, 0x5f, 0xa2, 0xfd, 0x15, 0xda, 0xcf, 0xd1, 0x8f, 0xb7,
-	0xfb, 0xac, 0xcf, 0x14, 0x32, 0x90, 0x51, 0x46, 0xf2, 0x7e, 0xe8, 0x50, 0x6d, 0xe5, 0x58, 0xf4,
-	0x08, 0xcc, 0x88, 0x0e, 0xfa, 0x91, 0x70, 0xf4, 0x9a, 0x5e, 0x37, 0x70, 0x9e, 0xc9, 0xfa, 0x19,
-	0x9b, 0x24, 0x44, 0x38, 0xa5, 0x9a, 0x5e, 0xbf, 0x8f, 0xf3, 0x4c, 0xd6, 0xbb, 0xd1, 0x74, 0x14,
-	0x73, 0xa7, 0x9c, 0xd5, 0xb3, 0x0c, 0x21, 0x30, 0x22, 0xc2, 0x23, 0xc7, 0xa8, 0xe9, 0x75, 0x1b,
-	0xab, 0x18, 0x1d, 0x43, 0x35, 0xa1, 0x82, 0xf4, 0x88, 0x20, 0x4e, 0xa5, 0xa6, 0xd7, 0xad, 0xc6,
-	0xae, 0xff, 0xcf, 0x86, 0xfd, 0x0f, 0x39, 0xbc, 0x69, 0x9c, 0xcf, 0x77, 0x34, 0xbc, 0xa2, 0x7b,
-	0xfb, 0x50, 0x2d, 0xfe, 0x43, 0x4f, 0xc1, 0x56, 0x1f, 0x3d, 0x95, 0x1f, 0xa1, 0xdc, 0xd1, 0x6b,
-	0xe5, 0xba, 0x8d, 0x2d, 0x55, 0x0b, 0x55, 0xc9, 0xfb, 0x66, 0x80, 0x5d, 0x1c, 0xf1, 0x58, 0xd0,
-	0x04, 0x85, 0x50, 0xe1, 0x82, 0x4d, 0xa8, 0x3a, 0xa5, 0xd5, 0x78, 0xfe, 0x9f, 0x3e, 0x0a, 0x6e,
-	0x4b, 0x72, 0xa4, 0x40, 0xa8, 0xe1, 0x4c, 0x00, 0x7d, 0x04, 0x63, 0x40, 0xd2, 0xa1, 0x1a, 0x8b,
-	0xd5, 0x08, 0xd6, 0x14, 0x3a, 0x7e, 0xdd, 0x7e, 0x2f, 0x75, 0x9a, 0xd5, 0xc5, 0x7c, 0xc7, 0x90,
-	0x59, 0xa8, 0x61, 0x25, 0x84, 0x3e, 0xc3, 0x06, 0xfd, 0x2a, 0xe8, 0x88, 0x0f, 0xd8, 0x48, 0x0d,
-	0xd5, 0x6a, 0x1c, 0xae, 0xa9, 0xfa, 0xb6, 0xe0, 0xc9, 0xd9, 0x84, 0x1a, 0xbe, 0x14, 0x42, 0x67,
-	0xb0, 0xb5, 0x4a, 0x4e, 0xc7, 0x64, 0x36, 0x64, 0xa4, 0xa7, 0xcc, 0xb1, 0x1a, 0x2f, 0xef, 0xaa,
-	0xfe, 0x29, 0xa3, 0x87, 0x1a, 0xde, 0xa4, 0xd7, 0x6a, 0xe8, 0x08, 0x4a, 0x71, 0x9a, 0xbb, 0xbb,
-	0xbf, 0xa6, 0xf0, 0x49, 0x5b, 0x8d, 0xc2, 0x5c, 0xcc, 0x77, 0x4a, 0x27, 0xed, 0x50, 0xc3, 0xa5,
-	0x38, 0x45, 0x47, 0x60, 0xf2, 0x6e, 0x44, 0x13, 0xe2, 0x98, 0x77, 0x12, 0x6b, 0x29, 0x52, 0xa8,
-	0xe1, 0x9c, 0xde, 0x34, 0xc1, 0x18, 0x08, 0x9a, 0x78, 0xbb, 0xb0, 0xf5, 0x97, 0x8d, 0xf2, 0x9a,
-	0x8e, 0x48, 0x92, 0x5d, 0x83, 0x0d, 0xac, 0x62, 0x6f, 0x08, 0x9b, 0xd7, 0x6d, 0x42, 0x9b, 0x50,
-	0x8e, 0xe9, 0x4c, 0xc1, 0x6c, 0x2c, 0x43, 0xb4, 0x0d, 0x95, 0x94, 0x0c, 0xa7, 0x54, 0x19, 0x6f,
-	0xe3, 0x2c, 0x41, 0x0e, 0xdc, 0x4b, 0xe9, 0x64, 0x65, 0x5d, 0x19, 0x17, 0xe9, 0x95, 0xc5, 0x92,
-	0x53, 0xaf, 0x14, 0x8b, 0xe5, 0xbd, 0x81, 0x87, 0x37, 0xda, 0x77, 0x53, 0x6b, 0xb7, 0x6d, 0xa1,
-	0x77, 0x08, 0xce, 0x6d, 0x2e, 0xc9, 0x96, 0x0a, 0xbf, 0xb3, 0xf6, 0x8b, 0xd4, 0x7b, 0x05, 0x0f,
-	0xfe, 0xb4, 0x60, 0xdd, 0x63, 0x7a, 0xcf, 0x2e, 0x99, 0xd9, 0xbc, 0x65, 0xb7, 0x31, 0x9d, 0x15,
-	0xcb, 0xa7, 0xe2, 0xe6, 0xbb, 0xf3, 0x85, 0xab, 0x5f, 0x2c, 0x5c, 0xfd, 0xd7, 0xc2, 0xd5, 0xbf,
-	0x2f, 0x5d, 0xed, 0x62, 0xe9, 0x6a, 0x3f, 0x97, 0xae, 0xf6, 0x65, 0xaf, 0x3f, 0x10, 0xd1, 0xb4,
-	0xe3, 0x77, 0x59, 0x12, 0xe4, 0x0f, 0x5c, 0xf6, 0xb3, 0xcf, 0x7b, 0xf1, 0x95, 0x67, 0x4e, 0xcc,
-	0xc6, 0x94, 0x77, 0x4c, 0xf5, 0x4e, 0xbd, 0xf8, 0x1d, 0x00, 0x00, 0xff, 0xff, 0x70, 0xb0, 0xd0,
-	0xad, 0x0c, 0x05, 0x00, 0x00,
+	// 513 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x53, 0xd1, 0x6e, 0xd3, 0x30,
+	0x14, 0x8d, 0xd7, 0xb4, 0x74, 0x37, 0x41, 0xea, 0xac, 0x81, 0x22, 0x24, 0xb2, 0x92, 0x97, 0xf5,
+	0x61, 0x4b, 0x58, 0x99, 0xc4, 0x33, 0x45, 0xa0, 0x54, 0x02, 0x81, 0x3c, 0xc4, 0x03, 0x2f, 0x93,
+	0xdb, 0x7a, 0x4d, 0xd4, 0x26, 0xae, 0x6a, 0xb7, 0xa2, 0x7f, 0xc1, 0xaf, 0xf0, 0x17, 0x7b, 0xdc,
+	0x23, 0x4f, 0x13, 0x6a, 0x3f, 0x80, 0x5f, 0x40, 0xb6, 0x93, 0x30, 0x8d, 0x0d, 0xb6, 0xa7, 0xde,
+	0x73, 0x7a, 0xee, 0xf1, 0xf5, 0xc9, 0x35, 0x1c, 0x0c, 0xb9, 0xc8, 0xb8, 0x88, 0x06, 0x54, 0xb0,
+	0x48, 0xe4, 0x74, 0x26, 0x12, 0x2e, 0x45, 0xb4, 0x3c, 0x1a, 0x30, 0x49, 0x8f, 0x2a, 0x26, 0x9c,
+	0xcd, 0xb9, 0xe4, 0xf8, 0xa9, 0x51, 0x87, 0x4a, 0x1d, 0x56, 0xea, 0xb0, 0x50, 0x3f, 0xd9, 0x1d,
+	0xf3, 0x31, 0xd7, 0xca, 0x48, 0x55, 0xa6, 0x29, 0xf8, 0x8e, 0xa0, 0x79, 0x52, 0x68, 0xf1, 0x63,
+	0x68, 0x24, 0x2c, 0x1d, 0x27, 0xd2, 0x43, 0x6d, 0xd4, 0xb1, 0x49, 0x81, 0x14, 0x7f, 0xc6, 0xe7,
+	0x19, 0x95, 0xde, 0x56, 0x1b, 0x75, 0x1e, 0x92, 0x02, 0x29, 0x7e, 0x98, 0x2c, 0xf2, 0x89, 0xf0,
+	0x6a, 0x86, 0x37, 0x08, 0x63, 0xb0, 0x13, 0x2a, 0x12, 0xcf, 0x6e, 0xa3, 0x8e, 0x4b, 0x74, 0x8d,
+	0xfb, 0xd0, 0xcc, 0x98, 0xa4, 0x23, 0x2a, 0xa9, 0x57, 0x6f, 0xa3, 0x8e, 0xd3, 0xdd, 0x0f, 0xff,
+	0x39, 0x70, 0xf8, 0xbe, 0x90, 0xf7, 0xec, 0xf3, 0xcb, 0x3d, 0x8b, 0x54, 0xed, 0xc1, 0x21, 0x34,
+	0xcb, 0xff, 0xf0, 0x33, 0x70, 0xf5, 0xa1, 0xa7, 0xea, 0x10, 0x26, 0x3c, 0xd4, 0xae, 0x75, 0x5c,
+	0xe2, 0x68, 0x2e, 0xd6, 0x54, 0xf0, 0x6b, 0x0b, 0xdc, 0xf2, 0x8a, 0x7d, 0xc9, 0x32, 0x1c, 0x43,
+	0x5d, 0x48, 0x3e, 0x67, 0xfa, 0x96, 0x4e, 0xf7, 0xf9, 0x7f, 0xe6, 0x28, 0x7b, 0x4f, 0x54, 0x8f,
+	0x32, 0x88, 0x2d, 0x62, 0x0c, 0xf0, 0x07, 0xb0, 0x53, 0xba, 0x9c, 0xea, 0x58, 0x9c, 0x6e, 0x74,
+	0x47, 0xa3, 0xfe, 0xab, 0xcf, 0xef, 0x94, 0x4f, 0xaf, 0xb9, 0xbe, 0xdc, 0xb3, 0x15, 0x8a, 0x2d,
+	0xa2, 0x8d, 0xf0, 0x27, 0xd8, 0x66, 0x5f, 0x25, 0xcb, 0x45, 0xca, 0x73, 0x1d, 0xaa, 0xd3, 0x3d,
+	0xbe, 0xa3, 0xeb, 0x9b, 0xb2, 0x4f, 0x65, 0x13, 0x5b, 0xe4, 0x8f, 0x11, 0x3e, 0x83, 0x9d, 0x0a,
+	0x9c, 0xce, 0xe8, 0x6a, 0xca, 0xe9, 0x48, 0x7f, 0x1c, 0xa7, 0xfb, 0xf2, 0xbe, 0xee, 0x1f, 0x4d,
+	0x7b, 0x6c, 0x91, 0x16, 0xbb, 0xc6, 0xf5, 0x1a, 0x60, 0xa7, 0x92, 0x65, 0xc1, 0x3e, 0xec, 0xfc,
+	0x15, 0x9a, 0x5a, 0x8a, 0x9c, 0x66, 0x26, 0xf4, 0x6d, 0xa2, 0xeb, 0x60, 0x0a, 0xad, 0xeb, 0xa1,
+	0xe0, 0x16, 0xd4, 0x26, 0x6c, 0xa5, 0x65, 0x2e, 0x51, 0x25, 0xde, 0x85, 0xfa, 0x92, 0x4e, 0x17,
+	0x4c, 0xc7, 0xec, 0x12, 0x03, 0xb0, 0x07, 0x0f, 0x96, 0x6c, 0x5e, 0x05, 0x55, 0x23, 0x25, 0xbc,
+	0xb2, 0xc6, 0xea, 0x8e, 0xf5, 0x72, 0x8d, 0x83, 0xd7, 0xf0, 0xe8, 0xc6, 0xb0, 0x6e, 0x1a, 0xed,
+	0xb6, 0x9d, 0x0f, 0x8e, 0xc1, 0xbb, 0x2d, 0x13, 0x35, 0x52, 0x99, 0xae, 0x19, 0xbf, 0x84, 0xbd,
+	0xb7, 0xe7, 0x6b, 0x1f, 0x5d, 0xac, 0x7d, 0xf4, 0x73, 0xed, 0xa3, 0x6f, 0x1b, 0xdf, 0xba, 0xd8,
+	0xf8, 0xd6, 0x8f, 0x8d, 0x6f, 0x7d, 0x39, 0x18, 0xa7, 0x32, 0x59, 0x0c, 0xc2, 0x21, 0xcf, 0xa2,
+	0xe2, 0xb9, 0x9b, 0x9f, 0x43, 0x31, 0x9a, 0x5c, 0x79, 0xf4, 0x72, 0x35, 0x63, 0x62, 0xd0, 0xd0,
+	0xaf, 0xf6, 0xc5, 0xef, 0x00, 0x00, 0x00, 0xff, 0xff, 0x6c, 0x90, 0xf8, 0x1f, 0x1a, 0x04, 0x00,
+	0x00,
 }
 
 func (m *Snapshot) Marshal() (dAtA []byte, err error) {
@@ -871,48 +725,6 @@ func (m *SnapshotItem_ExtensionPayload) MarshalToSizedBuffer(dAtA []byte) (int,
 	}
 	return len(dAtA) - i, nil
 }
-func (m *SnapshotItem_KV) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *SnapshotItem_KV) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	if m.KV != nil {
-		{
-			size, err := m.KV.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintSnapshot(dAtA, i, uint64(size))
-		}
-		i--
-		dAtA[i] = 0x2a
-	}
-	return len(dAtA) - i, nil
-}
-func (m *SnapshotItem_Schema) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *SnapshotItem_Schema) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	if m.Schema != nil {
-		{
-			size, err := m.Schema.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintSnapshot(dAtA, i, uint64(size))
-		}
-		i--
-		dAtA[i] = 0x32
-	}
-	return len(dAtA) - i, nil
-}
 func (m *SnapshotStoreItem) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -1055,75 +867,6 @@ func (m *SnapshotExtensionPayload) MarshalToSizedBuffer(dAtA []byte) (int, error
 	return len(dAtA) - i, nil
 }
 
-func (m *SnapshotKVItem) 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 *SnapshotKVItem) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *SnapshotKVItem) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if len(m.Value) > 0 {
-		i -= len(m.Value)
-		copy(dAtA[i:], m.Value)
-		i = encodeVarintSnapshot(dAtA, i, uint64(len(m.Value)))
-		i--
-		dAtA[i] = 0x12
-	}
-	if len(m.Key) > 0 {
-		i -= len(m.Key)
-		copy(dAtA[i:], m.Key)
-		i = encodeVarintSnapshot(dAtA, i, uint64(len(m.Key)))
-		i--
-		dAtA[i] = 0xa
-	}
-	return len(dAtA) - i, nil
-}
-
-func (m *SnapshotSchema) 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 *SnapshotSchema) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *SnapshotSchema) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if len(m.Keys) > 0 {
-		for iNdEx := len(m.Keys) - 1; iNdEx >= 0; iNdEx-- {
-			i -= len(m.Keys[iNdEx])
-			copy(dAtA[i:], m.Keys[iNdEx])
-			i = encodeVarintSnapshot(dAtA, i, uint64(len(m.Keys[iNdEx])))
-			i--
-			dAtA[i] = 0xa
-		}
-	}
-	return len(dAtA) - i, nil
-}
-
 func encodeVarintSnapshot(dAtA []byte, offset int, v uint64) int {
 	offset -= sovSnapshot(v)
 	base := offset
@@ -1234,30 +977,6 @@ func (m *SnapshotItem_ExtensionPayload) Size() (n int) {
 	}
 	return n
 }
-func (m *SnapshotItem_KV) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if m.KV != nil {
-		l = m.KV.Size()
-		n += 1 + l + sovSnapshot(uint64(l))
-	}
-	return n
-}
-func (m *SnapshotItem_Schema) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if m.Schema != nil {
-		l = m.Schema.Size()
-		n += 1 + l + sovSnapshot(uint64(l))
-	}
-	return n
-}
 func (m *SnapshotStoreItem) Size() (n int) {
 	if m == nil {
 		return 0
@@ -1323,38 +1042,6 @@ func (m *SnapshotExtensionPayload) Size() (n int) {
 	return n
 }
 
-func (m *SnapshotKVItem) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	l = len(m.Key)
-	if l > 0 {
-		n += 1 + l + sovSnapshot(uint64(l))
-	}
-	l = len(m.Value)
-	if l > 0 {
-		n += 1 + l + sovSnapshot(uint64(l))
-	}
-	return n
-}
-
-func (m *SnapshotSchema) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if len(m.Keys) > 0 {
-		for _, b := range m.Keys {
-			l = len(b)
-			n += 1 + l + sovSnapshot(uint64(l))
-		}
-	}
-	return n
-}
-
 func sovSnapshot(x uint64) (n int) {
 	return (math_bits.Len64(x|1) + 6) / 7
 }
@@ -1786,76 +1473,6 @@ func (m *SnapshotItem) Unmarshal(dAtA []byte) error {
 			}
 			m.Item = &SnapshotItem_ExtensionPayload{v}
 			iNdEx = postIndex
-		case 5:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field KV", wireType)
-			}
-			var msglen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowSnapshot
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				msglen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if msglen < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			postIndex := iNdEx + msglen
-			if postIndex < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			v := &SnapshotKVItem{}
-			if err := v.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
-				return err
-			}
-			m.Item = &SnapshotItem_KV{v}
-			iNdEx = postIndex
-		case 6:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Schema", wireType)
-			}
-			var msglen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowSnapshot
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				msglen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if msglen < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			postIndex := iNdEx + msglen
-			if postIndex < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			v := &SnapshotSchema{}
-			if err := v.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
-				return err
-			}
-			m.Item = &SnapshotItem_Schema{v}
-			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipSnapshot(dAtA[iNdEx:])
@@ -2300,206 +1917,6 @@ func (m *SnapshotExtensionPayload) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *SnapshotKVItem) 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 ErrIntOverflowSnapshot
-			}
-			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: SnapshotKVItem: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: SnapshotKVItem: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Key", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowSnapshot
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Key = append(m.Key[:0], dAtA[iNdEx:postIndex]...)
-			if m.Key == nil {
-				m.Key = []byte{}
-			}
-			iNdEx = postIndex
-		case 2:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Value", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowSnapshot
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Value = append(m.Value[:0], dAtA[iNdEx:postIndex]...)
-			if m.Value == nil {
-				m.Value = []byte{}
-			}
-			iNdEx = postIndex
-		default:
-			iNdEx = preIndex
-			skippy, err := skipSnapshot(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if (skippy < 0) || (iNdEx+skippy) < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
-func (m *SnapshotSchema) 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 ErrIntOverflowSnapshot
-			}
-			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: SnapshotSchema: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: SnapshotSchema: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Keys", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowSnapshot
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Keys = append(m.Keys, make([]byte, postIndex-iNdEx))
-			copy(m.Keys[len(m.Keys)-1], dAtA[iNdEx:postIndex])
-			iNdEx = postIndex
-		default:
-			iNdEx = preIndex
-			skippy, err := skipSnapshot(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if (skippy < 0) || (iNdEx+skippy) < 0 {
-				return ErrInvalidLengthSnapshot
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
 func skipSnapshot(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/snapshots/types/snapshotter.go b/snapshots/types/snapshotter.go
index 1641042a62339..f747920d13ad0 100644
--- a/snapshots/types/snapshotter.go
+++ b/snapshots/types/snapshotter.go
@@ -11,31 +11,17 @@ type Snapshotter interface {
 	// Snapshot writes snapshot items into the protobuf writer.
 	Snapshot(height uint64, protoWriter protoio.Writer) error
 
-	// PruneSnapshotHeight prunes the given height according to the prune strategy.
-	// If PruneNothing, this is a no-op.
-	// If other strategy, this height is persisted until it is
-	// less than <current height> - KeepRecent and <current height> % Interval == 0
-	PruneSnapshotHeight(height int64)
-
-	// SetSnapshotInterval sets the interval at which the snapshots are taken.
-	// It is used by the store that implements the Snapshotter interface
-	// to determine which heights to retain until after the snapshot is complete.
-	SetSnapshotInterval(snapshotInterval uint64)
-
-	// Restore restores a state snapshot, taking the reader of protobuf message stream as input.
+	// Restore restores a state snapshot from the protobuf items read from the reader.
+	// If the ready channel is non-nil, it returns a ready signal (by being closed) once the
+	// restorer is ready to accept chunks.
 	Restore(height uint64, format uint32, protoReader protoio.Reader) (SnapshotItem, error)
 }
 
-// ExtensionPayloadReader read extension payloads,
-// it returns io.EOF when reached either end of stream or the extension boundaries.
-type ExtensionPayloadReader = func() ([]byte, error)
-
-// ExtensionPayloadWriter is a helper to write extension payloads to underlying stream.
-type ExtensionPayloadWriter = func([]byte) error
-
 // ExtensionSnapshotter is an extension Snapshotter that is appended to the snapshot stream.
 // ExtensionSnapshotter has an unique name and manages it's own internal formats.
 type ExtensionSnapshotter interface {
+	Snapshotter
+
 	// SnapshotName returns the name of snapshotter, it should be unique in the manager.
 	SnapshotName() string
 
@@ -46,11 +32,4 @@ type ExtensionSnapshotter interface {
 
 	// SupportedFormats returns a list of formats it can restore from.
 	SupportedFormats() []uint32
-
-	// SnapshotExtension writes extension payloads into the underlying protobuf stream.
-	SnapshotExtension(height uint64, payloadWriter ExtensionPayloadWriter) error
-
-	// RestoreExtension restores an extension state snapshot,
-	// the payload reader returns `io.EOF` when reached the extension boundaries.
-	RestoreExtension(height uint64, format uint32, payloadReader ExtensionPayloadReader) error
 }
diff --git a/snapshots/types/util.go b/snapshots/types/util.go
index e2d4949bf5c2e..125ea6fb4610d 100644
--- a/snapshots/types/util.go
+++ b/snapshots/types/util.go
@@ -4,12 +4,12 @@ import (
 	protoio "github.com/gogo/protobuf/io"
 )
 
-// WriteExtensionPayload writes an extension payload for current extension snapshotter.
-func WriteExtensionPayload(protoWriter protoio.Writer, payload []byte) error {
+// WriteExtensionItem writes an item payload for current extention snapshotter.
+func WriteExtensionItem(protoWriter protoio.Writer, item []byte) error {
 	return protoWriter.WriteMsg(&SnapshotItem{
 		Item: &SnapshotItem_ExtensionPayload{
 			ExtensionPayload: &SnapshotExtensionPayload{
-				Payload: payload,
+				Payload: item,
 			},
 		},
 	})
diff --git a/store/rootmulti/snapshot_test.go b/store/rootmulti/snapshot_test.go
index bad1603da7c91..cbdb481a05f7f 100644
--- a/store/rootmulti/snapshot_test.go
+++ b/store/rootmulti/snapshot_test.go
@@ -18,12 +18,11 @@ import (
 	"github.com/cosmos/cosmos-sdk/store/iavl"
 	"github.com/cosmos/cosmos-sdk/store/rootmulti"
 	"github.com/cosmos/cosmos-sdk/store/types"
-	"github.com/tendermint/tendermint/libs/log"
 	dbm "github.com/tendermint/tm-db"
 )
 
 func newMultiStoreWithGeneratedData(db dbm.DB, stores uint8, storeKeys uint64) *rootmulti.Store {
-	multiStore := rootmulti.NewStore(db, log.NewNopLogger())
+	multiStore := rootmulti.NewStore(db)
 	r := rand.New(rand.NewSource(49872768940)) // Fixed seed for deterministic tests
 
 	keys := []*types.KVStoreKey{}
@@ -55,7 +54,7 @@ func newMultiStoreWithGeneratedData(db dbm.DB, stores uint8, storeKeys uint64) *
 }
 
 func newMultiStoreWithMixedMounts(db dbm.DB) *rootmulti.Store {
-	store := rootmulti.NewStore(db, log.NewNopLogger())
+	store := rootmulti.NewStore(db)
 	store.MountStoreWithDB(types.NewKVStoreKey("iavl1"), types.StoreTypeIAVL, nil)
 	store.MountStoreWithDB(types.NewKVStoreKey("iavl2"), types.StoreTypeIAVL, nil)
 	store.MountStoreWithDB(types.NewKVStoreKey("iavl3"), types.StoreTypeIAVL, nil)
@@ -211,8 +210,7 @@ func TestMultistoreSnapshotRestore(t *testing.T) {
 	require.Equal(t, *dummyExtensionItem.GetExtension(), *nextItem.GetExtension())
 
 	assert.Equal(t, source.LastCommitID(), target.LastCommitID())
-	for _, key := range source.StoreKeysByName() {
-		sourceStore := source.GetStoreByName(key.Name()).(types.CommitKVStore)
+	for key, sourceStore := range source.GetStores() {
 		targetStore := target.GetStoreByName(key.Name()).(types.CommitKVStore)
 		switch sourceStore.GetStoreType() {
 		case types.StoreTypeTransient:
@@ -235,8 +233,8 @@ func benchmarkMultistoreSnapshot(b *testing.B, stores uint8, storeKeys uint64) {
 	b.StartTimer()
 
 	for i := 0; i < b.N; i++ {
-		target := rootmulti.NewStore(dbm.NewMemDB(), log.NewNopLogger())
-		for _, key := range source.StoreKeysByName() {
+		target := rootmulti.NewStore(dbm.NewMemDB())
+		for key := range source.GetStores() {
 			target.MountStoreWithDB(key, types.StoreTypeIAVL, nil)
 		}
 		err := target.LoadLatestVersion()
@@ -270,8 +268,8 @@ func benchmarkMultistoreSnapshotRestore(b *testing.B, stores uint8, storeKeys ui
 	b.StartTimer()
 
 	for i := 0; i < b.N; i++ {
-		target := rootmulti.NewStore(dbm.NewMemDB(), log.NewNopLogger())
-		for _, key := range source.StoreKeysByName() {
+		target := rootmulti.NewStore(dbm.NewMemDB())
+		for key := range source.GetStores() {
 			target.MountStoreWithDB(key, types.StoreTypeIAVL, nil)
 		}
 		err := target.LoadLatestVersion()
diff --git a/store/rootmulti/store.go b/store/rootmulti/store.go
index 10d5de50b6d05..53cb6853cc23f 100644
--- a/store/rootmulti/store.go
+++ b/store/rootmulti/store.go
@@ -1,6 +1,7 @@
 package rootmulti
 
 import (
+	"encoding/binary"
 	"fmt"
 	"io"
 	"math"
@@ -13,11 +14,8 @@ import (
 	gogotypes "github.com/gogo/protobuf/types"
 	"github.com/pkg/errors"
 	abci "github.com/tendermint/tendermint/abci/types"
-	"github.com/tendermint/tendermint/libs/log"
 	dbm "github.com/tendermint/tm-db"
 
-	"github.com/cosmos/cosmos-sdk/pruning"
-	pruningtypes "github.com/cosmos/cosmos-sdk/pruning/types"
 	snapshottypes "github.com/cosmos/cosmos-sdk/snapshots/types"
 	"github.com/cosmos/cosmos-sdk/store/cachemulti"
 	"github.com/cosmos/cosmos-sdk/store/dbadapter"
@@ -32,6 +30,7 @@ import (
 
 const (
 	latestVersionKey = "s/latest"
+	pruneHeightsKey  = "s/pruneheights"
 	commitInfoKeyFmt = "s/%d" // s/<version>
 )
 
@@ -40,16 +39,15 @@ const (
 // the CommitMultiStore interface.
 type Store struct {
 	db             dbm.DB
-	logger         log.Logger
 	lastCommitInfo *types.CommitInfo
-	pruningManager *pruning.Manager
+	pruningOpts    types.PruningOptions
 	iavlCacheSize  int
 	storesParams   map[types.StoreKey]storeParams
 	stores         map[types.StoreKey]types.CommitKVStore
 	keysByName     map[string]types.StoreKey
 	lazyLoading    bool
+	pruneHeights   []int64
 	initialVersion int64
-	removalMap     map[types.StoreKey]bool
 
 	traceWriter       io.Writer
 	traceContext      types.TraceContext
@@ -69,40 +67,29 @@ var (
 // store will be created with a PruneNothing pruning strategy by default. After
 // a store is created, KVStores must be mounted and finally LoadLatestVersion or
 // LoadVersion must be called.
-func NewStore(db dbm.DB, logger log.Logger) *Store {
+func NewStore(db dbm.DB) *Store {
 	return &Store{
-		db:             db,
-		logger:         logger,
-		iavlCacheSize:  iavl.DefaultIAVLCacheSize,
-		storesParams:   make(map[types.StoreKey]storeParams),
-		stores:         make(map[types.StoreKey]types.CommitKVStore),
-		keysByName:     make(map[string]types.StoreKey),
-		listeners:      make(map[types.StoreKey][]types.WriteListener),
-		removalMap:     make(map[types.StoreKey]bool),
-		pruningManager: pruning.NewManager(db, logger),
+		db:            db,
+		pruningOpts:   types.PruneNothing,
+		iavlCacheSize: iavl.DefaultIAVLCacheSize,
+		storesParams:  make(map[types.StoreKey]storeParams),
+		stores:        make(map[types.StoreKey]types.CommitKVStore),
+		keysByName:    make(map[string]types.StoreKey),
+		pruneHeights:  make([]int64, 0),
+		listeners:     make(map[types.StoreKey][]types.WriteListener),
 	}
 }
 
 // GetPruning fetches the pruning strategy from the root store.
-func (rs *Store) GetPruning() pruningtypes.PruningOptions {
-	return rs.pruningManager.GetOptions()
+func (rs *Store) GetPruning() types.PruningOptions {
+	return rs.pruningOpts
 }
 
 // SetPruning sets the pruning strategy on the root store and all the sub-stores.
 // Note, calling SetPruning on the root store prior to LoadVersion or
 // LoadLatestVersion performs a no-op as the stores aren't mounted yet.
-func (rs *Store) SetPruning(pruningOpts pruningtypes.PruningOptions) {
-	rs.pruningManager.SetOptions(pruningOpts)
-}
-
-// SetSnapshotInterval sets the interval at which the snapshots are taken.
-// It is used by the store to determine which heights to retain until after the snapshot is complete.
-func (rs *Store) SetSnapshotInterval(snapshotInterval uint64) {
-	rs.pruningManager.SetSnapshotInterval(snapshotInterval)
-}
-
-func (rs *Store) SetIAVLCacheSize(cacheSize int) {
-	rs.iavlCacheSize = cacheSize
+func (rs *Store) SetPruning(pruningOpts types.PruningOptions) {
+	rs.pruningOpts = pruningOpts
 }
 
 func (rs *Store) SetIAVLCacheSize(cacheSize int) {
@@ -159,9 +146,9 @@ func (rs *Store) GetCommitKVStore(key types.StoreKey) types.CommitKVStore {
 	return rs.stores[key]
 }
 
-// StoreKeysByName returns mapping storeNames -> StoreKeys
-func (rs *Store) StoreKeysByName() map[string]types.StoreKey {
-	return rs.keysByName
+// GetStores returns mounted stores
+func (rs *Store) GetStores() map[types.StoreKey]types.CommitKVStore {
+	return rs.stores
 }
 
 // LoadLatestVersionAndUpgrade implements CommitMultiStore
@@ -206,7 +193,7 @@ func (rs *Store) loadVersion(ver int64, upgrades *types.StoreUpgrades) error {
 	}
 
 	// load each Store (note this doesn't panic on unmounted keys now)
-	newStores := make(map[types.StoreKey]types.CommitKVStore)
+	var newStores = make(map[types.StoreKey]types.CommitKVStore)
 
 	storesKeys := make([]types.StoreKey, 0, len(rs.storesParams))
 
@@ -243,10 +230,9 @@ func (rs *Store) loadVersion(ver int64, upgrades *types.StoreUpgrades) error {
 			if err := deleteKVStore(store.(types.KVStore)); err != nil {
 				return errors.Wrapf(err, "failed to delete store %s", key.Name())
 			}
-			rs.removalMap[key] = true
 		} else if oldName := upgrades.RenamedFrom(key.Name()); oldName != "" {
 			// handle renames specially
-			// make an unregistered key to satisfy loadCommitStore params
+			// make an unregistered key to satify loadCommitStore params
 			oldKey := types.NewKVStoreKey(oldName)
 			oldParams := storeParams
 			oldParams.key = oldKey
@@ -261,11 +247,6 @@ func (rs *Store) loadVersion(ver int64, upgrades *types.StoreUpgrades) error {
 			if err := moveKVStoreData(oldStore.(types.KVStore), store.(types.KVStore)); err != nil {
 				return errors.Wrapf(err, "failed to move store %s -> %s", oldName, key.Name())
 			}
-
-			// add the old key so its deletion is committed
-			newStores[oldKey] = oldStore
-			// this will ensure it's not perpetually stored in commitInfo
-			rs.removalMap[oldKey] = true
 		}
 	}
 
@@ -273,8 +254,9 @@ func (rs *Store) loadVersion(ver int64, upgrades *types.StoreUpgrades) error {
 	rs.stores = newStores
 
 	// load any pruned heights we missed from disk to be pruned on the next run
-	if err := rs.pruningManager.LoadPruningHeights(rs.db); err != nil {
-		return err
+	ph, err := getPruningHeights(rs.db)
+	if err == nil && len(ph) > 0 {
+		rs.pruneHeights = ph
 	}
 
 	return nil
@@ -319,14 +301,6 @@ func moveKVStoreData(oldDB types.KVStore, newDB types.KVStore) error {
 	return deleteKVStore(oldDB)
 }
 
-// PruneSnapshotHeight prunes the given height according to the prune strategy.
-// If PruneNothing, this is a no-op.
-// If other strategy, this height is persisted until it is
-// less than <current height> - KeepRecent and <current height> % Interval == 0
-func (rs *Store) PruneSnapshotHeight(height int64) {
-	rs.pruningManager.HandleHeightSnapshot(height)
-}
-
 // SetInterBlockCache sets the Store's internal inter-block (persistent) cache.
 // When this is defined, all CommitKVStores will be wrapped with their respective
 // inter-block cache.
@@ -348,7 +322,13 @@ func (rs *Store) SetTracer(w io.Writer) types.MultiStore {
 func (rs *Store) SetTracingContext(tc types.TraceContext) types.MultiStore {
 	rs.traceContextMutex.Lock()
 	defer rs.traceContextMutex.Unlock()
-	rs.traceContext = rs.traceContext.Merge(tc)
+	if rs.traceContext != nil {
+		for k, v := range tc {
+			rs.traceContext[k] = v
+		}
+	} else {
+		rs.traceContext = tc
+	}
 
 	return rs
 }
@@ -409,6 +389,7 @@ func (rs *Store) Commit() types.CommitID {
 		// This case means that no commit has been made in the store, we
 		// start from initialVersion.
 		version = rs.initialVersion
+
 	} else {
 		// This case can means two things:
 		// - either there was already a previous commit in the store, in which
@@ -419,30 +400,59 @@ func (rs *Store) Commit() types.CommitID {
 		version = previousHeight + 1
 	}
 
-	rs.lastCommitInfo = commitStores(version, rs.stores, rs.removalMap)
-	defer rs.flushMetadata(rs.db, version, rs.lastCommitInfo)
+	rs.lastCommitInfo = commitStores(version, rs.stores)
 
-	// remove remnants of removed stores
-	for sk := range rs.removalMap {
-		if _, ok := rs.stores[sk]; ok {
-			delete(rs.stores, sk)
-			delete(rs.storesParams, sk)
-			delete(rs.keysByName, sk.Name())
+	// Determine if pruneHeight height needs to be added to the list of heights to
+	// be pruned, where pruneHeight = (commitHeight - 1) - KeepRecent.
+	if rs.pruningOpts.Interval > 0 && int64(rs.pruningOpts.KeepRecent) < previousHeight {
+		pruneHeight := previousHeight - int64(rs.pruningOpts.KeepRecent)
+		// We consider this height to be pruned iff:
+		//
+		// - KeepEvery is zero as that means that all heights should be pruned.
+		// - KeepEvery % (height - KeepRecent) != 0 as that means the height is not
+		// a 'snapshot' height.
+		if rs.pruningOpts.KeepEvery == 0 || pruneHeight%int64(rs.pruningOpts.KeepEvery) != 0 {
+			rs.pruneHeights = append(rs.pruneHeights, pruneHeight)
 		}
 	}
-	// reset the removalMap
-	rs.removalMap = make(map[types.StoreKey]bool)
 
-	if err := rs.handlePruning(version); err != nil {
-		panic(err)
+	// batch prune if the current height is a pruning interval height
+	if rs.pruningOpts.Interval > 0 && version%int64(rs.pruningOpts.Interval) == 0 {
+		rs.pruneStores()
 	}
 
+	flushMetadata(rs.db, version, rs.lastCommitInfo, rs.pruneHeights)
+
 	return types.CommitID{
 		Version: version,
 		Hash:    rs.lastCommitInfo.Hash(),
 	}
 }
 
+// pruneStores will batch delete a list of heights from each mounted sub-store.
+// Afterwards, pruneHeights is reset.
+func (rs *Store) pruneStores() {
+	if len(rs.pruneHeights) == 0 {
+		return
+	}
+
+	for key, store := range rs.stores {
+		if store.GetStoreType() == types.StoreTypeIAVL {
+			// If the store is wrapped with an inter-block cache, we must first unwrap
+			// it to get the underlying IAVL store.
+			store = rs.GetCommitKVStore(key)
+
+			if err := store.(*iavl.Store).DeleteVersions(rs.pruneHeights...); err != nil {
+				if errCause := errors.Cause(err); errCause != nil && errCause != iavltree.ErrVersionDoesNotExist {
+					panic(err)
+				}
+			}
+		}
+	}
+
+	rs.pruneHeights = make([]int64, 0)
+}
+
 // CacheWrap implements CacheWrapper/Store/CommitStore.
 func (rs *Store) CacheWrap() types.CacheWrap {
 	return rs.CacheMultiStore().(types.CacheWrap)
@@ -536,51 +546,7 @@ func (rs *Store) GetKVStore(key types.StoreKey) types.KVStore {
 	return store
 }
 
-func (rs *Store) handlePruning(version int64) error {
-	rs.pruningManager.HandleHeight(version - 1) // we should never prune the current version.
-	if !rs.pruningManager.ShouldPruneAtHeight(version) {
-		return nil
-	}
-	rs.logger.Info("prune start", "height", version)
-	defer rs.logger.Info("prune end", "height", version)
-	return rs.pruneStores()
-}
-
-func (rs *Store) pruneStores() error {
-	pruningHeights, err := rs.pruningManager.GetFlushAndResetPruningHeights()
-	if err != nil {
-		return err
-	}
-
-	if len(pruningHeights) == 0 {
-		rs.logger.Debug("pruning skipped; no heights to prune")
-		return nil
-	}
-
-	rs.logger.Debug("pruning heights", "heights", pruningHeights)
-
-	for key, store := range rs.stores {
-		// If the store is wrapped with an inter-block cache, we must first unwrap
-		// it to get the underlying IAVL store.
-		if store.GetStoreType() != types.StoreTypeIAVL {
-			continue
-		}
-
-		store = rs.GetCommitKVStore(key)
-
-		err := store.(*iavl.Store).DeleteVersions(pruningHeights...)
-		if err == nil {
-			continue
-		}
-
-		if errCause := errors.Cause(err); errCause != nil && errCause != iavltree.ErrVersionDoesNotExist {
-			return err
-		}
-	}
-	return nil
-}
-
-// getStoreByName performs a lookup of a StoreKey given a store name typically
+// GetStoreByName performs a lookup of a StoreKey given a store name typically
 // provided in a path. The StoreKey is then used to perform a lookup and return
 // a Store. If the Store is wrapped in an inter-block cache, it will be unwrapped
 // prior to being returned. If the StoreKey does not exist, nil is returned.
@@ -601,17 +567,17 @@ func (rs *Store) Query(req abci.RequestQuery) abci.ResponseQuery {
 	path := req.Path
 	storeName, subpath, err := parsePath(path)
 	if err != nil {
-		return sdkerrors.QueryResult(err, false)
+		return sdkerrors.QueryResult(err)
 	}
 
 	store := rs.GetStoreByName(storeName)
 	if store == nil {
-		return sdkerrors.QueryResult(sdkerrors.Wrapf(sdkerrors.ErrUnknownRequest, "no such store: %s", storeName), false)
+		return sdkerrors.QueryResult(sdkerrors.Wrapf(sdkerrors.ErrUnknownRequest, "no such store: %s", storeName))
 	}
 
 	queryable, ok := store.(types.Queryable)
 	if !ok {
-		return sdkerrors.QueryResult(sdkerrors.Wrapf(sdkerrors.ErrUnknownRequest, "store %s (type %T) doesn't support queries", storeName, store), false)
+		return sdkerrors.QueryResult(sdkerrors.Wrapf(sdkerrors.ErrUnknownRequest, "store %s (type %T) doesn't support queries", storeName, store))
 	}
 
 	// trim the path and make the query
@@ -623,7 +589,7 @@ func (rs *Store) Query(req abci.RequestQuery) abci.ResponseQuery {
 	}
 
 	if res.ProofOps == nil || len(res.ProofOps.Ops) == 0 {
-		return sdkerrors.QueryResult(sdkerrors.Wrap(sdkerrors.ErrInvalidRequest, "proof is unexpectedly empty; ensure height has not been pruned"), false)
+		return sdkerrors.QueryResult(sdkerrors.Wrap(sdkerrors.ErrInvalidRequest, "proof is unexpectedly empty; ensure height has not been pruned"))
 	}
 
 	// If the request's height is the latest height we've committed, then utilize
@@ -636,7 +602,7 @@ func (rs *Store) Query(req abci.RequestQuery) abci.ResponseQuery {
 	} else {
 		commitInfo, err = getCommitInfo(rs.db, res.Height)
 		if err != nil {
-			return sdkerrors.QueryResult(err, false)
+			return sdkerrors.QueryResult(err)
 		}
 	}
 
@@ -658,7 +624,7 @@ func (rs *Store) SetInitialVersion(version int64) error {
 			// If the store is wrapped with an inter-block cache, we must first unwrap
 			// it to get the underlying IAVL store.
 			store = rs.GetCommitKVStore(key)
-			store.(types.StoreWithInitialVersion).SetInitialVersion(version)
+			store.(*iavl.Store).SetInitialVersion(version)
 		}
 	}
 
@@ -693,7 +659,7 @@ func (rs *Store) Snapshot(height uint64, protoWriter protoio.Writer) error {
 	if height == 0 {
 		return sdkerrors.Wrap(sdkerrors.ErrLogic, "cannot snapshot height 0")
 	}
-	if height > uint64(getLatestVersion(rs.db)) {
+	if height > uint64(rs.LastCommitID().Version) {
 		return sdkerrors.Wrapf(sdkerrors.ErrLogic, "cannot snapshot future height %v", height)
 	}
 
@@ -846,7 +812,7 @@ loop:
 		importer.Close()
 	}
 
-	rs.flushMetadata(rs.db, int64(height), rs.buildCommitInfo(int64(height)))
+	flushMetadata(rs.db, int64(height), rs.buildCommitInfo(int64(height)), []int64{})
 	return snapshotItem, rs.LoadLatestVersion()
 }
 
@@ -869,9 +835,9 @@ func (rs *Store) loadCommitStoreFromParams(key types.StoreKey, id types.CommitID
 		var err error
 
 		if params.initialVersion == 0 {
-			store, err = iavl.LoadStore(db, rs.logger, key, id, rs.lazyLoading, rs.iavlCacheSize)
+			store, err = iavl.LoadStore(db, id, rs.lazyLoading, rs.iavlCacheSize)
 		} else {
-			store, err = iavl.LoadStoreWithInitialVersion(db, rs.logger, key, id, rs.lazyLoading, params.initialVersion, rs.iavlCacheSize)
+			store, err = iavl.LoadStoreWithInitialVersion(db, id, rs.lazyLoading, params.initialVersion, rs.iavlCacheSize)
 		}
 
 		if err != nil {
@@ -911,17 +877,8 @@ func (rs *Store) loadCommitStoreFromParams(key types.StoreKey, id types.CommitID
 }
 
 func (rs *Store) buildCommitInfo(version int64) *types.CommitInfo {
-	keys := make([]types.StoreKey, 0, len(rs.stores))
-	for key := range rs.stores {
-		keys = append(keys, key)
-	}
-	sort.Slice(keys, func(i, j int) bool {
-		return keys[i].Name() < keys[j].Name()
-	})
-
 	storeInfos := []types.StoreInfo{}
-	for _, key := range keys {
-		store := rs.stores[key]
+	for key, store := range rs.stores {
 		if store.GetStoreType() == types.StoreTypeTransient {
 			continue
 		}
@@ -936,51 +893,6 @@ func (rs *Store) buildCommitInfo(version int64) *types.CommitInfo {
 	}
 }
 
-// RollbackToVersion delete the versions after `target` and update the latest version.
-func (rs *Store) RollbackToVersion(target int64) int64 {
-	if target < 0 {
-		panic("Negative rollback target")
-	}
-	current := getLatestVersion(rs.db)
-	if target >= current {
-		return current
-	}
-	for ; current > target; current-- {
-		rs.pruningManager.HandleHeight(current)
-	}
-	if err := rs.pruneStores(); err != nil {
-		panic(err)
-	}
-
-	// update latest height
-	bz, err := gogotypes.StdInt64Marshal(current)
-	if err != nil {
-		panic(err)
-	}
-
-	rs.db.Set([]byte(latestVersionKey), bz)
-	return current
-}
-
-func (rs *Store) flushMetadata(db dbm.DB, version int64, cInfo *types.CommitInfo) {
-	rs.logger.Debug("flushing metadata", "height", version)
-	batch := db.NewBatch()
-	defer batch.Close()
-
-	if cInfo != nil {
-		flushCommitInfo(batch, version, cInfo)
-	} else {
-		rs.logger.Debug("commitInfo is nil, not flushed", "height", version)
-	}
-
-	flushLatestVersion(batch, version)
-
-	if err := batch.WriteSync(); err != nil {
-		panic(fmt.Errorf("error on batch write %w", err))
-	}
-	rs.logger.Debug("flushing metadata finished", "height", version)
-}
-
 type storeParams struct {
 	key            types.StoreKey
 	db             dbm.DB
@@ -1006,7 +918,7 @@ func getLatestVersion(db dbm.DB) int64 {
 }
 
 // Commits each store and returns a new commitInfo.
-func commitStores(version int64, storeMap map[types.StoreKey]types.CommitKVStore, removalMap map[types.StoreKey]bool) *types.CommitInfo {
+func commitStores(version int64, storeMap map[types.StoreKey]types.CommitKVStore) *types.CommitInfo {
 	storeInfos := make([]types.StoreInfo, 0, len(storeMap))
 
 	for key, store := range storeMap {
@@ -1016,18 +928,12 @@ func commitStores(version int64, storeMap map[types.StoreKey]types.CommitKVStore
 			continue
 		}
 
-		if !removalMap[key] {
-			si := types.StoreInfo{}
-			si.Name = key.Name()
-			si.CommitId = commitID
-			storeInfos = append(storeInfos, si)
-		}
+		si := types.StoreInfo{}
+		si.Name = key.Name()
+		si.CommitId = commitID
+		storeInfos = append(storeInfos, si)
 	}
 
-	sort.SliceStable(storeInfos, func(i, j int) bool {
-		return strings.Compare(storeInfos[i].Name, storeInfos[j].Name) < 0
-	})
-
 	return &types.CommitInfo{
 		Version:    version,
 		StoreInfos: storeInfos,
@@ -1053,7 +959,7 @@ func getCommitInfo(db dbm.DB, ver int64) (*types.CommitInfo, error) {
 	return cInfo, nil
 }
 
-func flushCommitInfo(batch dbm.Batch, version int64, cInfo *types.CommitInfo) {
+func setCommitInfo(batch dbm.Batch, version int64, cInfo *types.CommitInfo) {
 	bz, err := cInfo.Marshal()
 	if err != nil {
 		panic(err)
@@ -1063,7 +969,7 @@ func flushCommitInfo(batch dbm.Batch, version int64, cInfo *types.CommitInfo) {
 	batch.Set([]byte(cInfoKey), bz)
 }
 
-func flushLatestVersion(batch dbm.Batch, version int64) {
+func setLatestVersion(batch dbm.Batch, version int64) {
 	bz, err := gogotypes.StdInt64Marshal(version)
 	if err != nil {
 		panic(err)
@@ -1071,3 +977,47 @@ func flushLatestVersion(batch dbm.Batch, version int64) {
 
 	batch.Set([]byte(latestVersionKey), bz)
 }
+
+func setPruningHeights(batch dbm.Batch, pruneHeights []int64) {
+	bz := make([]byte, 0)
+	for _, ph := range pruneHeights {
+		buf := make([]byte, 8)
+		binary.BigEndian.PutUint64(buf, uint64(ph))
+		bz = append(bz, buf...)
+	}
+
+	batch.Set([]byte(pruneHeightsKey), bz)
+}
+
+func getPruningHeights(db dbm.DB) ([]int64, error) {
+	bz, err := db.Get([]byte(pruneHeightsKey))
+	if err != nil {
+		return nil, fmt.Errorf("failed to get pruned heights: %w", err)
+	}
+	if len(bz) == 0 {
+		return nil, errors.New("no pruned heights found")
+	}
+
+	prunedHeights := make([]int64, len(bz)/8)
+	i, offset := 0, 0
+	for offset < len(bz) {
+		prunedHeights[i] = int64(binary.BigEndian.Uint64(bz[offset : offset+8]))
+		i++
+		offset += 8
+	}
+
+	return prunedHeights, nil
+}
+
+func flushMetadata(db dbm.DB, version int64, cInfo *types.CommitInfo, pruneHeights []int64) {
+	batch := db.NewBatch()
+	defer batch.Close()
+
+	setCommitInfo(batch, version, cInfo)
+	setLatestVersion(batch, version)
+	setPruningHeights(batch, pruneHeights)
+
+	if err := batch.Write(); err != nil {
+		panic(fmt.Errorf("error on batch write %w", err))
+	}
+}
diff --git a/store/rootmulti/store_test.go b/store/rootmulti/store_test.go
index ccd58b153feb3..e2fe5835727f7 100644
--- a/store/rootmulti/store_test.go
+++ b/store/rootmulti/store_test.go
@@ -8,12 +8,10 @@ import (
 
 	"github.com/stretchr/testify/require"
 	abci "github.com/tendermint/tendermint/abci/types"
-	"github.com/tendermint/tendermint/libs/log"
 	dbm "github.com/tendermint/tm-db"
 
 	"github.com/cosmos/cosmos-sdk/codec"
 	codecTypes "github.com/cosmos/cosmos-sdk/codec/types"
-	pruningtypes "github.com/cosmos/cosmos-sdk/pruning/types"
 	"github.com/cosmos/cosmos-sdk/store/cachemulti"
 	"github.com/cosmos/cosmos-sdk/store/iavl"
 	sdkmaps "github.com/cosmos/cosmos-sdk/store/internal/maps"
@@ -24,13 +22,13 @@ import (
 
 func TestStoreType(t *testing.T) {
 	db := dbm.NewMemDB()
-	store := NewStore(db, log.NewNopLogger())
+	store := NewStore(db)
 	store.MountStoreWithDB(types.NewKVStoreKey("store1"), types.StoreTypeIAVL, db)
 }
 
 func TestGetCommitKVStore(t *testing.T) {
 	var db dbm.DB = dbm.NewMemDB()
-	ms := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningDefault))
+	ms := newMultiStoreWithMounts(db, types.PruneDefault)
 	err := ms.LoadLatestVersion()
 	require.Nil(t, err)
 
@@ -47,7 +45,7 @@ func TestGetCommitKVStore(t *testing.T) {
 
 func TestStoreMount(t *testing.T) {
 	db := dbm.NewMemDB()
-	store := NewStore(db, log.NewNopLogger())
+	store := NewStore(db)
 
 	key1 := types.NewKVStoreKey("store1")
 	key2 := types.NewKVStoreKey("store2")
@@ -63,7 +61,7 @@ func TestStoreMount(t *testing.T) {
 
 func TestCacheMultiStore(t *testing.T) {
 	var db dbm.DB = dbm.NewMemDB()
-	ms := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	ms := newMultiStoreWithMounts(db, types.PruneNothing)
 
 	cacheMulti := ms.CacheMultiStore()
 	require.IsType(t, cachemulti.Store{}, cacheMulti)
@@ -71,7 +69,7 @@ func TestCacheMultiStore(t *testing.T) {
 
 func TestCacheMultiStoreWithVersion(t *testing.T) {
 	var db dbm.DB = dbm.NewMemDB()
-	ms := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	ms := newMultiStoreWithMounts(db, types.PruneNothing)
 	err := ms.LoadLatestVersion()
 	require.Nil(t, err)
 
@@ -108,7 +106,7 @@ func TestCacheMultiStoreWithVersion(t *testing.T) {
 
 func TestHashStableWithEmptyCommit(t *testing.T) {
 	var db dbm.DB = dbm.NewMemDB()
-	ms := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	ms := newMultiStoreWithMounts(db, types.PruneNothing)
 	err := ms.LoadLatestVersion()
 	require.Nil(t, err)
 
@@ -132,7 +130,7 @@ func TestHashStableWithEmptyCommit(t *testing.T) {
 
 func TestMultistoreCommitLoad(t *testing.T) {
 	var db dbm.DB = dbm.NewMemDB()
-	store := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	store := newMultiStoreWithMounts(db, types.PruneNothing)
 	err := store.LoadLatestVersion()
 	require.Nil(t, err)
 
@@ -157,7 +155,7 @@ func TestMultistoreCommitLoad(t *testing.T) {
 	}
 
 	// Load the latest multistore again and check version.
-	store = newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	store = newMultiStoreWithMounts(db, types.PruneNothing)
 	err = store.LoadLatestVersion()
 	require.Nil(t, err)
 	commitID = getExpectedCommitID(store, nCommits)
@@ -170,7 +168,7 @@ func TestMultistoreCommitLoad(t *testing.T) {
 
 	// Load an older multistore and check version.
 	ver := nCommits - 1
-	store = newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	store = newMultiStoreWithMounts(db, types.PruneNothing)
 	err = store.LoadVersion(ver)
 	require.Nil(t, err)
 	commitID = getExpectedCommitID(store, ver)
@@ -179,7 +177,7 @@ func TestMultistoreCommitLoad(t *testing.T) {
 
 func TestMultistoreLoadWithUpgrade(t *testing.T) {
 	var db dbm.DB = dbm.NewMemDB()
-	store := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	store := newMultiStoreWithMounts(db, types.PruneNothing)
 	err := store.LoadLatestVersion()
 	require.Nil(t, err)
 
@@ -214,7 +212,7 @@ func TestMultistoreLoadWithUpgrade(t *testing.T) {
 	checkContains(t, ci.StoreInfos, []string{"store1", "store2", "store3"})
 
 	// Load without changes and make sure it is sensible
-	store = newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	store = newMultiStoreWithMounts(db, types.PruneNothing)
 
 	err = store.LoadLatestVersion()
 	require.Nil(t, err)
@@ -227,7 +225,7 @@ func TestMultistoreLoadWithUpgrade(t *testing.T) {
 	require.Equal(t, v2, s2.Get(k2))
 
 	// now, let's load with upgrades...
-	restore, upgrades := newMultiStoreWithModifiedMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	restore, upgrades := newMultiStoreWithModifiedMounts(db, types.PruneNothing)
 	err = restore.LoadLatestVersionAndUpgrade(upgrades)
 	require.Nil(t, err)
 
@@ -272,7 +270,7 @@ func TestMultistoreLoadWithUpgrade(t *testing.T) {
 	migratedID := restore.Commit()
 	require.Equal(t, migratedID.Version, int64(2))
 
-	reload, _ := newMultiStoreWithModifiedMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	reload, _ := newMultiStoreWithModifiedMounts(db, types.PruneNothing)
 	err = reload.LoadLatestVersion()
 	require.Nil(t, err)
 	require.Equal(t, migratedID, reload.LastCommitID())
@@ -294,8 +292,8 @@ func TestMultistoreLoadWithUpgrade(t *testing.T) {
 	ci, err = getCommitInfo(db, 2)
 	require.NoError(t, err)
 	require.Equal(t, int64(2), ci.Version)
-	require.Equal(t, 3, len(ci.StoreInfos), ci.StoreInfos)
-	checkContains(t, ci.StoreInfos, []string{"store1", "restore2", "store4"})
+	require.Equal(t, 4, len(ci.StoreInfos), ci.StoreInfos)
+	checkContains(t, ci.StoreInfos, []string{"store1", "restore2", "store3", "store4"})
 }
 
 func TestParsePath(t *testing.T) {
@@ -316,11 +314,16 @@ func TestParsePath(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, substore, "bang")
 	require.Equal(t, subsubpath, "/baz")
+
 }
 
 func TestMultiStoreRestart(t *testing.T) {
 	db := dbm.NewMemDB()
-	pruning := pruningtypes.NewCustomPruningOptions(2, 1)
+	pruning := types.PruningOptions{
+		KeepRecent: 2,
+		KeepEvery:  3,
+		Interval:   1,
+	}
 	multi := newMultiStoreWithMounts(db, pruning)
 	err := multi.LoadLatestVersion()
 	require.Nil(t, err)
@@ -399,7 +402,7 @@ func TestMultiStoreRestart(t *testing.T) {
 
 func TestMultiStoreQuery(t *testing.T) {
 	db := dbm.NewMemDB()
-	multi := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	multi := newMultiStoreWithMounts(db, types.PruneNothing)
 	err := multi.LoadLatestVersion()
 	require.Nil(t, err)
 
@@ -426,7 +429,7 @@ func TestMultiStoreQuery(t *testing.T) {
 	ver := cid.Version
 
 	// Reload multistore from database
-	multi = newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	multi = newMultiStoreWithMounts(db, types.PruneNothing)
 	err = multi.LoadLatestVersion()
 	require.Nil(t, err)
 
@@ -471,15 +474,15 @@ func TestMultiStore_Pruning(t *testing.T) {
 	testCases := []struct {
 		name        string
 		numVersions int64
-		po          pruningtypes.PruningOptions
+		po          types.PruningOptions
 		deleted     []int64
 		saved       []int64
 	}{
-		{"prune nothing", 10, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing), nil, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}},
-		{"prune everything", 10, pruningtypes.NewPruningOptions(pruningtypes.PruningEverything), []int64{1, 2, 3, 4, 5, 6, 7, 8, 9}, []int64{10}},
-		{"prune some; no batch", 10, pruningtypes.NewCustomPruningOptions(2, 1), []int64{1, 2, 4, 5, 7}, []int64{3, 6, 8, 9, 10}},
-		{"prune some; small batch", 10, pruningtypes.NewCustomPruningOptions(2, 3), []int64{1, 2, 4, 5}, []int64{3, 6, 7, 8, 9, 10}},
-		{"prune some; large batch", 10, pruningtypes.NewCustomPruningOptions(2, 11), nil, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}},
+		{"prune nothing", 10, types.PruneNothing, nil, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}},
+		{"prune everything", 10, types.PruneEverything, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9}, []int64{10}},
+		{"prune some; no batch", 10, types.NewPruningOptions(2, 3, 1), []int64{1, 2, 4, 5, 7}, []int64{3, 6, 8, 9, 10}},
+		{"prune some; small batch", 10, types.NewPruningOptions(2, 3, 3), []int64{1, 2, 4, 5}, []int64{3, 6, 7, 8, 9, 10}},
+		{"prune some; large batch", 10, types.NewPruningOptions(2, 3, 11), nil, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}},
 	}
 
 	for _, tc := range testCases {
@@ -507,64 +510,9 @@ func TestMultiStore_Pruning(t *testing.T) {
 	}
 }
 
-func TestMultiStore_Pruning_SameHeightsTwice(t *testing.T) {
-	const (
-		numVersions int64  = 10
-		keepRecent  uint64 = 2
-		interval    uint64 = 10
-	)
-
-	expectedHeights := []int64{}
-	for i := int64(1); i < numVersions-int64(keepRecent); i++ {
-		expectedHeights = append(expectedHeights, i)
-	}
-
-	db := dbm.NewMemDB()
-
-	ms := newMultiStoreWithMounts(db, pruningtypes.NewCustomPruningOptions(keepRecent, interval))
-	require.NoError(t, ms.LoadLatestVersion())
-
-	var lastCommitInfo types.CommitID
-	for i := int64(0); i < numVersions; i++ {
-		lastCommitInfo = ms.Commit()
-	}
-
-	require.Equal(t, numVersions, lastCommitInfo.Version)
-
-	for v := int64(1); v < numVersions-int64(keepRecent); v++ {
-		err := ms.LoadVersion(v)
-		require.Error(t, err, "expected error when loading pruned height: %d", v)
-	}
-
-	for v := int64(numVersions - int64(keepRecent)); v < numVersions; v++ {
-		err := ms.LoadVersion(v)
-		require.NoError(t, err, "expected no error when loading height: %d", v)
-	}
-
-	// Get latest
-	err := ms.LoadVersion(numVersions - 1)
-	require.NoError(t, err)
-
-	// Ensure already pruned heights were loaded
-	heights, err := ms.pruningManager.GetFlushAndResetPruningHeights()
-	require.NoError(t, err)
-	require.Equal(t, expectedHeights, heights)
-
-	require.NoError(t, ms.pruningManager.LoadPruningHeights(db))
-
-	// Test pruning the same heights again
-	lastCommitInfo = ms.Commit()
-	require.Equal(t, numVersions, lastCommitInfo.Version)
-
-	// Ensure that can commit one more height with no panic
-	lastCommitInfo = ms.Commit()
-	require.Equal(t, numVersions+1, lastCommitInfo.Version)
-}
-
 func TestMultiStore_PruningRestart(t *testing.T) {
 	db := dbm.NewMemDB()
-	ms := newMultiStoreWithMounts(db, pruningtypes.NewCustomPruningOptions(2, 11))
-	ms.SetSnapshotInterval(3)
+	ms := newMultiStoreWithMounts(db, types.NewPruningOptions(2, 3, 11))
 	require.NoError(t, ms.LoadLatestVersion())
 
 	// Commit enough to build up heights to prune, where on the next block we should
@@ -576,30 +524,19 @@ func TestMultiStore_PruningRestart(t *testing.T) {
 	pruneHeights := []int64{1, 2, 4, 5, 7}
 
 	// ensure we've persisted the current batch of heights to prune to the store's DB
-	err := ms.pruningManager.LoadPruningHeights(ms.db)
-	require.NoError(t, err)
-
-	actualHeightsToPrune, err := ms.pruningManager.GetFlushAndResetPruningHeights()
+	ph, err := getPruningHeights(ms.db)
 	require.NoError(t, err)
-	require.Equal(t, len(pruneHeights), len(actualHeightsToPrune))
-	require.Equal(t, pruneHeights, actualHeightsToPrune)
+	require.Equal(t, pruneHeights, ph)
 
 	// "restart"
-	ms = newMultiStoreWithMounts(db, pruningtypes.NewCustomPruningOptions(2, 11))
-	ms.SetSnapshotInterval(3)
+	ms = newMultiStoreWithMounts(db, types.NewPruningOptions(2, 3, 11))
 	err = ms.LoadLatestVersion()
 	require.NoError(t, err)
-
-	actualHeightsToPrune, err = ms.pruningManager.GetFlushAndResetPruningHeights()
-	require.NoError(t, err)
-	require.Equal(t, pruneHeights, actualHeightsToPrune)
+	require.Equal(t, pruneHeights, ms.pruneHeights)
 
 	// commit one more block and ensure the heights have been pruned
 	ms.Commit()
-
-	actualHeightsToPrune, err = ms.pruningManager.GetFlushAndResetPruningHeights()
-	require.NoError(t, err)
-	require.Empty(t, actualHeightsToPrune)
+	require.Empty(t, ms.pruneHeights)
 
 	for _, v := range pruneHeights {
 		_, err := ms.CacheMultiStoreWithVersion(v)
@@ -609,7 +546,7 @@ func TestMultiStore_PruningRestart(t *testing.T) {
 
 func TestSetInitialVersion(t *testing.T) {
 	db := dbm.NewMemDB()
-	multi := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	multi := newMultiStoreWithMounts(db, types.PruneNothing)
 
 	require.NoError(t, multi.LoadLatestVersion())
 
@@ -627,7 +564,7 @@ func TestSetInitialVersion(t *testing.T) {
 
 func TestAddListenersAndListeningEnabled(t *testing.T) {
 	db := dbm.NewMemDB()
-	multi := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	multi := newMultiStoreWithMounts(db, types.PruneNothing)
 	testKey := types.NewKVStoreKey("listening_test_key")
 	enabled := multi.ListeningEnabled(testKey)
 	require.False(t, enabled)
@@ -658,7 +595,7 @@ var (
 func TestGetListenWrappedKVStore(t *testing.T) {
 	buf := new(bytes.Buffer)
 	var db dbm.DB = dbm.NewMemDB()
-	ms := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	ms := newMultiStoreWithMounts(db, types.PruneNothing)
 	ms.LoadLatestVersion()
 	mockListeners := []types.WriteListener{types.NewStoreKVPairWriteListener(buf, testMarshaller)}
 	ms.AddListeners(testStoreKey1, mockListeners)
@@ -701,7 +638,6 @@ func TestGetListenWrappedKVStore(t *testing.T) {
 		StoreKey: testStoreKey2.Name(),
 		Delete:   false,
 	})
-	require.NoError(t, err)
 	kvPairSet2Bytes := buf.Bytes()
 	buf.Reset()
 	require.Equal(t, expectedOutputKVPairSet2, kvPairSet2Bytes)
@@ -713,7 +649,6 @@ func TestGetListenWrappedKVStore(t *testing.T) {
 		StoreKey: testStoreKey2.Name(),
 		Delete:   true,
 	})
-	require.NoError(t, err)
 	kvPairDelete2Bytes := buf.Bytes()
 	buf.Reset()
 	require.Equal(t, expectedOutputKVPairDelete2, kvPairDelete2Bytes)
@@ -734,7 +669,7 @@ func TestGetListenWrappedKVStore(t *testing.T) {
 
 func TestCacheWraps(t *testing.T) {
 	db := dbm.NewMemDB()
-	multi := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	multi := newMultiStoreWithMounts(db, types.PruneNothing)
 
 	cacheWrapper := multi.CacheWrap()
 	require.IsType(t, cachemulti.Store{}, cacheWrapper)
@@ -748,7 +683,7 @@ func TestCacheWraps(t *testing.T) {
 
 func TestTraceConcurrency(t *testing.T) {
 	db := dbm.NewMemDB()
-	multi := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
+	multi := newMultiStoreWithMounts(db, types.PruneNothing)
 	err := multi.LoadLatestVersion()
 	require.NoError(t, err)
 
@@ -796,40 +731,6 @@ func TestTraceConcurrency(t *testing.T) {
 	stopW <- struct{}{}
 }
 
-func TestCommitOrdered(t *testing.T) {
-	var db dbm.DB = dbm.NewMemDB()
-	multi := newMultiStoreWithMounts(db, pruningtypes.NewPruningOptions(pruningtypes.PruningNothing))
-	err := multi.LoadLatestVersion()
-	require.Nil(t, err)
-
-	commitID := types.CommitID{}
-	checkStore(t, multi, commitID, commitID)
-
-	k, v := []byte("wind"), []byte("blows")
-	k2, v2 := []byte("water"), []byte("flows")
-	k3, v3 := []byte("fire"), []byte("burns")
-
-	store1 := multi.GetStoreByName("store1").(types.KVStore)
-	store1.Set(k, v)
-
-	store2 := multi.GetStoreByName("store2").(types.KVStore)
-	store2.Set(k2, v2)
-
-	store3 := multi.GetStoreByName("store3").(types.KVStore)
-	store3.Set(k3, v3)
-
-	typeID := multi.Commit()
-	require.Equal(t, int64(1), typeID.Version)
-
-	ci, err := getCommitInfo(db, 1)
-	require.NoError(t, err)
-	require.Equal(t, int64(1), ci.Version)
-	require.Equal(t, 3, len(ci.StoreInfos))
-	for i, s := range ci.StoreInfos {
-		require.Equal(t, s.Name, fmt.Sprintf("store%d", i+1))
-	}
-}
-
 //-----------------------------------------------------------------------
 // utils
 
@@ -839,9 +740,9 @@ var (
 	testStoreKey3 = types.NewKVStoreKey("store3")
 )
 
-func newMultiStoreWithMounts(db dbm.DB, pruningOpts pruningtypes.PruningOptions) *Store {
-	store := NewStore(db, log.NewNopLogger())
-	store.SetPruning(pruningOpts)
+func newMultiStoreWithMounts(db dbm.DB, pruningOpts types.PruningOptions) *Store {
+	store := NewStore(db)
+	store.pruningOpts = pruningOpts
 
 	store.MountStoreWithDB(testStoreKey1, types.StoreTypeIAVL, nil)
 	store.MountStoreWithDB(testStoreKey2, types.StoreTypeIAVL, nil)
@@ -850,9 +751,9 @@ func newMultiStoreWithMounts(db dbm.DB, pruningOpts pruningtypes.PruningOptions)
 	return store
 }
 
-func newMultiStoreWithModifiedMounts(db dbm.DB, pruningOpts pruningtypes.PruningOptions) (*Store, *types.StoreUpgrades) {
-	store := NewStore(db, log.NewNopLogger())
-	store.SetPruning(pruningOpts)
+func newMultiStoreWithModifiedMounts(db dbm.DB, pruningOpts types.PruningOptions) (*Store, *types.StoreUpgrades) {
+	store := NewStore(db)
+	store.pruningOpts = pruningOpts
 
 	store.MountStoreWithDB(types.NewKVStoreKey("store1"), types.StoreTypeIAVL, nil)
 	store.MountStoreWithDB(types.NewKVStoreKey("restore2"), types.StoreTypeIAVL, nil)